Improve the API significantly.

Author: nickelser

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.1.5
+
+- Improve the API.
+- Add scheduler test.
+
 ## 0.1.4
 
 - Fix several bugs related to time parsing.

README.md

@@ -15,34 +15,37 @@ gem 'zhong'
 ## Usage
 
 ```ruby
-r = Redis.new
-
-Zhong.schedule(redis: r) do |s|
-  s.category "stuff" do
-    s.every(5.seconds, "foo") { puts "foo" }
-    s.every(1.minute, "biz", at: ["**:26", "**:27"]) { puts "biz" }
-    s.every(1.week, "baz", at: ["mon 22:45", "wed 23:13"]) { puts "baz" }
-    s.every(10.seconds, "boom") { raise "fail" }
+Zhong.redis = Redis.new(ENV["ZHONG_REDIS_URL"])
+
+Zhong.schedule do
+  category "stuff" do
+    every 5.seconds, "foo" do
+      puts "foo"
+    end
+
+    every(1.minute, "biz", at: ["**:26", "**:27"]) { puts "biz" }
+    every(1.week, "baz", at: ["mon 22:45", "wed 23:13"]) { puts "baz" }
+    every(10.seconds, "boom") { raise "fail" }
   end
 
-  s.category "clutter" do
-    s.every(1.second, "compute", if: -> (t) { t.wday == 3 && rand < 0.5 }) do
+  category "clutter" do
+    every(1.second, "compute", if: -> (t) { t.wday == 3 && rand < 0.5 }) do
       puts "something happened on wednesday, maybe"
     end
   end
 
   # note: callbacks that return nil or false will cause event to not run
-  s.on(:before_tick) do
+  on(:before_tick) do
     puts "ding"
     true
   end
 
-  s.on(:after_tick) do
+  on(:after_tick) do
     puts "dong"
     true
   end
 
-  s.error_handler do |e, job|
+  error_handler do |e, job|
     puts "damn, #{job} messed up: #{e}"
   end
 end

bin/zhong

@@ -2,14 +2,18 @@
 
 STDERR.sync = STDOUT.sync = true
 
-require "bundler/setup"
 require "zhong"
 
-usage = "zhong <zhong.rb>"
-file = ARGV.shift || abort(usage)
+file = ARGV.shift or abort "zhong <path to zhong.rb>"
 
 file = "./#{file}" unless file.match(/^[\/.]/)
 
 require file
 
-Zhong.start
+begin
+  Zhong.start
+rescue => boom
+  STDERR.puts boom.message
+  STDERR.puts boom.backtrace.join("\n")
+  exit 1
+end

lib/zhong.rb

@@ -5,8 +5,6 @@
 
 require "zhong/version"
 
-require "zhong/util"
-
 require "zhong/at"
 require "zhong/every"
 
@@ -15,19 +13,32 @@
 
 module Zhong
   class << self
-    def schedule(**opts)
-      @scheduler = Scheduler.new(opts).tap do |s|
-        yield(s)
-      end
-    end
+    attr_writer :logger, :redis
+  end
 
-    def start
-      fail "You must run `Zhong.schedule` first" unless scheduler
-      scheduler.start
-    end
+  def self.schedule(&block)
+    scheduler.instance_eval(&block) if block_given?
+  end
+
+  def self.start
+    scheduler.start
+  end
 
-    def scheduler
-      @scheduler
+  def self.stop
+    scheduler.stop
+  end
+
+  def self.scheduler
+    @scheduler ||= Scheduler.new(logger: logger, redis: redis)
+  end
+
+  def self.logger
+    @logger ||= Logger.new(STDOUT).tap do |logger|
+      logger.formatter = -> (_, datetime, _, msg) { "#{datetime}: #{msg}\n" }
     end
   end
+
+  def self.redis
+    @redis ||= Redis.new(url: ENV["REDIS_URL"])
+  end
 end

lib/zhong/every.rb

@@ -14,7 +14,7 @@ class FailedToParse < StandardError; end
     def initialize(period)
       @period = period
 
-      fail ArgumentError unless valid?
+      fail "`every` must be >= 1 second" unless valid?
     end
 
     private def valid?

lib/zhong/job.rb

@@ -1,6 +1,6 @@
 module Zhong
   class Job
-    attr_reader :name, :category, :last_ran
+    attr_reader :name, :category, :last_ran, :logger
 
     def initialize(name, config = {}, &block)
       @name = name
@@ -17,8 +17,8 @@ def initialize(name, config = {}, &block)
       @redis = config[:redis]
       @tz = config[:tz]
       @if = config[:if]
-      @lock = Suo::Client::Redis.new(lock_key, client: @redis, stale_lock_expiration: config[:long_running_timeout])
-      @timeout = 5
+      @long_running_timeout = config[:long_running_timeout]
+      @running = false
 
       refresh_last_ran
     end
@@ -30,18 +30,13 @@ def run?(time = Time.now)
     def run(time = Time.now, error_handler = nil)
       return unless run?(time)
 
-      if running?
-        @logger.info "already running: #{self}"
-        return
-      end
-
-      @thread = nil
       locked = false
       errored = false
 
       begin
-        @lock.lock do
+        redis_lock.lock do
           locked = true
+          @running = true
 
           refresh_last_ran
 
@@ -50,44 +45,35 @@ def run(time = Time.now, error_handler = nil)
           break unless run?(time)
 
           if disabled?
-            @logger.info "disabled: #{self}"
+            logger.info "disabled: #{self}"
             break
           end
 
-          @logger.info "running: #{self}"
+          logger.info "running: #{self}"
 
           if @block
-            @thread = Thread.new do
-              begin
-                @block.call
-              rescue => boom
-                @logger.error "#{self} failed: #{boom}"
-                error_handler.call(boom, self) if error_handler
-              end
-
-              nil # do not retain thread return value
+            begin
+              @block.call
+            rescue => boom
+              logger.error "#{self} failed: #{boom}"
+              error_handler.call(boom, self) if error_handler
             end
           end
 
           ran!(time)
         end
       rescue Suo::LockClientError => boom
-        @logger.error "unable to run due to client error: #{boom}"
+        logger.error "unable to run due to client error: #{boom}"
         errored = true
       end
 
-      @logger.info "unable to acquire exclusive run lock: #{self}" if !locked && !errored
-    end
+      @running = false
 
-    def stop
-      return unless running?
-      Thread.new { @logger.error "killing #{self} due to stop" } # thread necessary due to trap context
-      @thread.join(@timeout)
-      @thread.kill
+      logger.info "unable to acquire exclusive run lock: #{self}" if !locked && !errored
     end
 
     def running?
-      @thread && @thread.alive?
+      @running
     end
 
     def refresh_last_ran
@@ -140,6 +126,10 @@ def ran!(time)
       @redis.set(last_ran_key, @last_ran.to_i)
     end
 
+    def redis_lock
+      @lock ||= Suo::Client::Redis.new(lock_key, client: @redis, stale_lock_expiration: @long_running_timeout)
+    end
+
     def last_ran_key
       "zhong:last_ran:#{self}"
     end

lib/zhong/scheduler.rb

@@ -5,15 +5,18 @@ class Scheduler
     DEFAULT_CONFIG = {
       timeout: 0.5,
       grace: 15.minutes,
-      long_running_timeout: 5.minutes
+      long_running_timeout: 5.minutes,
+      tz: nil
     }.freeze
 
     def initialize(config = {})
       @jobs = {}
       @callbacks = {}
       @config = DEFAULT_CONFIG.merge(config)
-      @logger = @config[:logger] ||= Util.default_logger
-      @redis = @config[:redis] ||= Redis.new(ENV["REDIS_URL"])
+      
+      @logger = @config[:logger]
+      @redis = @config[:redis]
+      @tz = @config[:tz]
     end
 
     def category(name)
@@ -28,8 +31,15 @@ def category(name)
 
     def every(period, name, opts = {}, &block)
       fail "must specify a period for #{name} (#{caller.first})" unless period
+
       job = Job.new(name, opts.merge(@config).merge(every: period, category: @category), &block)
-      add(job)
+
+      if jobs.key?(job.to_s)
+        @logger.error "duplicate job #{job}, skipping"
+        return
+      end
+      
+      @jobs[job.to_s] = job
     end
 
     def error_handler(&block)
@@ -45,6 +55,8 @@ def on(event, &block)
     def start
       @logger.info "starting at #{redis_time}"
 
+      @stop = false
+
       trap_signals
 
       loop do
@@ -57,20 +69,25 @@ def start
 
           fire_callbacks(:after_tick)
 
-          sleep_until_next_tick
+          sleep_until_next_second
         end
 
         break if @stop
       end
+
+      Thread.new { @logger.info "stopped" }
     end
 
     def stop
       Thread.new { @logger.error "stopping" } # thread necessary due to trap context
       @stop = true
-      jobs.values.each(&:stop)
-      Thread.new { @logger.info "stopped" }
     end
 
+    private
+
+    TRAPPED_SIGNALS = %w(QUIT INT TERM).freeze
+    private_constant :TRAPPED_SIGNALS
+
     def fire_callbacks(event, *args)
       @callbacks[event].to_a.all? { |h| h.call(*args) }
     end
@@ -87,35 +104,21 @@ def run_job(job, time = redis_time)
       fire_callbacks(:after_run, job, time)
     end
 
-    private
-
-    TRAPPED_SIGNALS = %w(QUIT INT TERM).freeze
-    private_constant :TRAPPED_SIGNALS
-
     def trap_signals
       TRAPPED_SIGNALS.each do |sig|
         Signal.trap(sig) { stop }
       end
     end
 
-    def add(job)
-      if @jobs.key?(job.to_s)
-        @logger.error "duplicate job #{job}, skipping"
-        return
-      end
-
-      @jobs[job.to_s] = job
-    end
-
-    def sleep_until_next_tick
+    def sleep_until_next_second
       GC.start
-      sleep(1.0 - Time.now.subsec + 0.001)
+      sleep(1.0 - Time.now.subsec + 0.0001)
     end
 
     def redis_time
       s, ms = @redis.time # returns [seconds since epoch, microseconds]
       now = Time.at(s + ms / (10**6))
-      config[:tz] ? now.in_time_zone(config[:tz]) : now
+      @tz ? now.in_time_zone(@tz) : now
     end
   end
 end

lib/zhong/util.rb

@@ -1,3 +1,3 @@
 module Zhong
-  VERSION = "0.1.4"
+  VERSION = "0.1.5"
 end