Class: PerfectSched::Backend::RDBCompatBackend

Inherits:
Object
  • Object
show all
Includes:
PerfectSched::BackendHelper
Defined in:
lib/perfectsched/backend/rdb_compat.rb

Defined Under Namespace

Classes: Token

Constant Summary collapse

MAX_RETRY = 
10
MAX_SELECT_ROW = 
4

Instance Attribute Summary collapse

Attributes included from PerfectSched::BackendHelper

#client

Instance Method Summary collapse

Methods included from PerfectSched::BackendHelper

#close

Constructor Details

#initialize(client, config) ⇒ RDBCompatBackend

Returns a new instance of RDBCompatBackend.



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 28

def initialize(client, config)
  super

  require 'sequel'
  url = config[:url]
  unless url
    raise ConfigError, "url option is required for the rdb_compat backend"
  end

  @table = config[:table]
  unless @table
    raise ConfigError, "table option is required for the rdb_compat backend"
  end

  #password = config[:password]
  #user = config[:user]

  case url.split('//',2)[0].to_s
  when /sqlite/i
    @db = Sequel.connect(url, :max_connections=>1)
  when /mysql/i
    require 'uri'

    uri = URI.parse(url)
    options = {
      user: uri.user,
      password: uri.password,
      host: uri.host,
      port: uri.port ? uri.port.to_i : 3306
    }
    options[:sslca] = config[:sslca] if config[:sslca]

    db_name = uri.path.split('/')[1]
    @db = Sequel.mysql2(db_name, options)
  else
    raise ConfigError, "'sqlite' and 'mysql' are supported"
  end

  @mutex = Mutex.new

  connect {
    # connection test
  }
end

Instance Attribute Details

#db ⇒ Object (readonly)

Returns the value of attribute db.



75
76
77
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 75

def db
  @db
end

Instance Method Details

#acquire(alive_time, max_acquire, options) ⇒ Object 



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 164

def acquire(alive_time, max_acquire, options)
  now = (options[:now] || Time.now).to_i
  next_timeout = now + alive_time

  connect {
    while true
      rows = 0
      @db.fetch("SELECT id, timeout, next_time, cron, delay, data, timezone FROM `#{@table}` WHERE timeout <= ? ORDER BY timeout ASC LIMIT #{MAX_SELECT_ROW};", now) {|row|

        n = @db["UPDATE `#{@table}` SET timeout=? WHERE id=? AND timeout=?;", next_timeout, row[:id], row[:timeout]].update
        if n > 0
          scheduled_time = row[:next_time]
          attributes = create_attributes(row)
          task_token = Token.new(row[:id], row[:next_time], attributes[:cron], attributes[:delay], attributes[:timezone])
          task = Task.new(@client, row[:id], attributes, scheduled_time, task_token)
          return [task]
        end

        rows += 1
      }
      if rows < MAX_SELECT_ROW
        return nil
      end
    end
  }
end

#add(key, type, cron, delay, timezone, data, next_time, next_run_time, options) ⇒ Object 



115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 115

def add(key, type, cron, delay, timezone, data, next_time, next_run_time, options)
  data = data ? data.dup : {}
  data['type'] = type
  connect {
    begin
      n = @db["INSERT INTO `#{@table}` (id, timeout, next_time, cron, delay, data, timezone) VALUES (?, ?, ?, ?, ?, ?, ?);", key, next_run_time, next_time, cron, delay, data.to_json, timezone].insert
      return Schedule.new(@client, key)
    rescue Sequel::DatabaseError
      raise IdempotentAlreadyExistsError, "schedule key=#{key} already exists"
    end
  }
end

#delete(key, options) ⇒ Object 



128
129
130
131
132
133
134
135
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 128

def delete(key, options)
  connect {
    n = @db["DELETE FROM `#{@table}` WHERE id=?;", key].delete
    if n <= 0
      raise IdempotentNotFoundError, "schedule key=#{key} does no exist"
    end
  }
end

#finish(task_token, options) ⇒ Object 



212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 212

def finish(task_token, options)
  row_id = task_token.row_id
  scheduled_time = task_token.scheduled_time
  next_time = PerfectSched.next_time(task_token.cron, scheduled_time, task_token.timezone)
  next_run_time = next_time + task_token.delay

  connect {
    n = @db["UPDATE `#{@table}` SET timeout=?, next_time=? WHERE id=? AND next_time=?;", next_run_time, next_time, row_id, scheduled_time].update
    if n <= 0
      raise IdempotentAlreadyFinishedError, "task time=#{Time.at(scheduled_time).utc} is already finished"
    end
  }
end

#get_schedule_metadata(key, options = {}) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 94

def get_schedule_metadata(key, options={})
  connect {
    row = @db.fetch("SELECT id, timeout, next_time, cron, delay, data, timezone FROM `#{@table}` WHERE id=? LIMIT 1", key).first
    unless row
      raise NotFoundError, "schedule key=#{key} does not exist"
    end
    attributes = create_attributes(row)
    return ScheduleWithMetadata.new(@client, key, attributes)
  }
end

#heartbeat(task_token, alive_time, options) ⇒ Object 



191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 191

def heartbeat(task_token, alive_time, options)
  now = (options[:now] || Time.now).to_i
  row_id = task_token.row_id
  scheduled_time = task_token.scheduled_time
  next_run_time = now + alive_time

  connect {
    n = @db["UPDATE `#{@table}` SET timeout=? WHERE id=? AND next_time=?;", next_run_time, row_id, scheduled_time].update
    if n <= 0  # TODO fix
      row = @db.fetch("SELECT id, timeout, next_time FROM `#{@table}` WHERE id=? AND next_time=? LIMIT 1", row_id, scheduled_time).first
      if row == nil
        raise PreemptedError, "task #{task_token.inspect} does not exist or preempted."
      elsif row[:timeout] == next_run_time
        # ok
      else
        raise PreemptedError, "task time=#{Time.at(scheduled_time).utc} is preempted"
      end
    end
  }
end

#init_database(options) ⇒ Object 



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 77

def init_database(options)
  sql = %[
    CREATE TABLE IF NOT EXISTS `#{@table}` (
      id VARCHAR(255) NOT NULL,
      timeout INT NOT NULL,
      next_time INT NOT NULL,
      cron VARCHAR(128) NOT NULL,
      delay INT NOT NULL,
      data LONGBLOB NOT NULL,
      timezone VARCHAR(255) NULL,
      PRIMARY KEY (id)
    );]
  connect {
    @db.run sql
  }
end

#list(options, &block) ⇒ Object 



105
106
107
108
109
110
111
112
113
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 105

def list(options, &block)
  connect {
    @db.fetch("SELECT id, timeout, next_time, cron, delay, data, timezone FROM `#{@table}` ORDER BY timeout ASC") {|row|
      attributes = create_attributes(row)
      sched = ScheduleWithMetadata.new(@client, row[:id], attributes)
      yield sched
    }
  }
end

#modify(key, options) ⇒ Object 



137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 137

def modify(key, options)
  ks = []
  vs = []
  [:cron, :delay, :timezone].each {|k|
    # TODO type and data are not supported
    if v = options[k]
      ks << k
      vs << v
    end
  }
  return nil if ks.empty?

  sql = "UPDATE `#{@table}` SET "
  sql << ks.map {|k| "#{k}=?" }.join(', ')
  sql << " WHERE id=?"

  args = [sql].concat(vs)
  args << key

  connect {
    n = @db[*args].update
    if n <= 0
      raise NotFoundError, "schedule key=#{key} does not exist"
    end
  }
end