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_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.



27
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
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 27

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.



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

def db
  @db
end

Instance Method Details

#acquire(alive_time, max_acquire, options) ⇒ Object 



163
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
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 163

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 



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

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 



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

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 



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

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 



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

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 ScheduleMetadata.new(@client, key, attributes)
  }
end

#heartbeat(task_token, alive_time, options) ⇒ Object 



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

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 



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

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

#list(options, &block) ⇒ Object 



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

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 



136
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
 
# File 'lib/perfectsched/backend/rdb_compat.rb', line 136

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