Class: Envoi::Aspera::WatchService::WatchFolder

Inherits:
Object
  • Object
show all
Defined in:
lib/envoi/aspera/watch_service/watch_folder.rb

Defined Under Namespace

Classes: State, Subscription

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(definition, state = nil, client = nil) ⇒ WatchFolder

Returns a new instance of WatchFolder.



161
162
163
164
165
166
167
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 161

def initialize(definition, state = nil, client = nil)
  @definition = definition
  @state      = state || State.new
  @client     = @watcher = client || Client.new
  @poll_interval = definition.fetch('poll_interval', 15)
  process_definition(definition)
end

Instance Attribute Details

#client ⇒ Object

Returns the value of attribute client.



158
159
160
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 158

def client
  @client
end

#definition ⇒ Object

Returns the value of attribute definition.



158
159
160
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 158

def definition
  @definition
end

#last_poll_time ⇒ Object

Returns the value of attribute last_poll_time.



158
159
160
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 158

def last_poll_time
  @last_poll_time
end

#poll_interval ⇒ Object

Returns the value of attribute poll_interval.



158
159
160
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 158

def poll_interval
  @poll_interval
end

#previous_poll_time ⇒ Object

Returns the value of attribute previous_poll_time.



158
159
160
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 158

def previous_poll_time
  @previous_poll_time
end

#state ⇒ Object

Returns the value of attribute state.



158
159
160
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 158

def state
  @state
end

#watcher ⇒ Object

Returns the value of attribute watcher.



158
159
160
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 158

def watcher
  @watcher
end

Class Method Details

.process_watch_folder(watch_folder, &block) ⇒ Object 



294
295
296
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 294

def self.process_watch_folder(watch_folder, &block)
  watch_folder.poll(&block) if (Time.now - watch_folder.last_poll_time) >= watch_folder.poll_interval
end

.process_watch_folder_def(watch_folder_def) ⇒ Object 



285
286
287
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 285

def self.process_watch_folder_def(watch_folder_def)
  new(watch_folder_def)
end

.process_watch_folder_defs(watch_folder_defs) ⇒ Object 



289
290
291
292
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 289

def self.process_watch_folder_defs(watch_folder_defs)


end

.process_watch_folders(watch_folders, &block) ⇒ Object 



298
299
300
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 298

def self.process_watch_folders(watch_folders, &block)
  watch_folders.each { |watch_folder| process_watch_folder(watch_folder, &block) }
end

.run(watch_folders, poll_interval = 15, &block) ⇒ Object 



306
307
308
309
310
311
312
313
314
315
316
317
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 306

def self.run(watch_folders, poll_interval = 15, &block)
  puts 'Starting...'
  loop do
    begin
      run_once(watch_folders, &block)
      sleep 1
    rescue SystemExit, Interrupt
      break
    end
  end
  puts 'Exiting...'
end

.run_once(watch_folders, &block) ⇒ Object 



302
303
304
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 302

def self.run_once(watch_folders, &block)
  process_watch_folders(watch_folders, &block)
end

Instance Method Details

#events(from, to) ⇒ Object 



189
190
191
192
193
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 189

def events(from, to)
  state.subscription.client.subscription_snapshot_differential(subscription_id: state.subscription['identifier'],
                                                               from:            from,
                                                               to:              to)
end

#logger ⇒ Object 



169
170
171
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 169

def logger
  client.logger
end

#path ⇒ Object 



173
174
175
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 173

def path
  definition['path']
end

#poll {|_self| ... } ⇒ Object

Yields:

  • (_self)

Yield Parameters:



195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 195

def poll
  subscription              = state.subscription
  subscription.resubscribe
  previous_snapshot_version = subscription.snapshots.keys.last

  # Delete other snapshots
  subscription.snapshots.delete_if { |p, e| p != previous_snapshot_version } if previous_snapshot_version

  snapshot = subscription.snapshot_create
  @previous_poll_time = @last_poll_time
  @last_poll_time = Time.now

  no_change = snapshot.version == previous_snapshot_version
  if !no_change
    events = snapshot.differential(previous_snapshot_version)
    logger.debug { "Events: #{events}" }
  end

  _known_path_map = state.known_path_map.dup

  logger.debug { "Subscription ID: #{subscription['identifier']}" }
  logger.debug { "Previous Snapshot: #{previous_snapshot_version}" }
  logger.debug { "Current  Snapshot: #{snapshot.version}" }
  logger.debug { "Known Paths: #{_known_path_map.keys}" }

  current_path_map = snapshot.entries_by_path
  current_paths    = current_path_map.keys

  new_path_map = current_path_map.dup

  deleted_path_map  = {}
  stable_path_map   = {}
  unstable_path_map = {}
  _known_path_map.delete_if do |p, e|
    deleted = !no_change && !current_paths.include?(p)
    if deleted
      logger.debug { "DELETED '#{p}'" }
      deleted_path_map[p] = e
    else
      new_path_map.delete(p) # The file is not new, so remove it from the new list

      new_entry     = current_path_map[p]
      previous_stat = e.stat.values_at('mtime', 'size')
      new_stat      = new_entry.stat.values_at('mtime', 'size')

      if no_change || new_stat === previous_stat
        logger.debug { "UNCHANGED '#{p}' '#{new_stat}' == '#{previous_stat}'" }
        stable_poll_count     = e[:stable_poll_count] || 0
        stable_poll_count     += 1
        e[:stable_poll_count] = stable_poll_count
        stable_path_map[p]    = e
      else
        logger.debug { "MODIFIED '#{p}' '#{new_stat}' != '#{previous_stat}'" }
        e[:stable_poll_count] = 0
        unstable_path_map[p]  = new_entry
      end
    end
    deleted
  end

  new_path_map.each do |p, e|
    logger.debug { "CREATED '#{p}' #{e.stat.values_at('mtime', 'size')}" }
    e[:stable_poll_count] = 0
    unstable_path_map[p]  = e.dup
  end

  _known_path_map.merge! stable_path_map
  _known_path_map.merge! unstable_path_map unless no_change
  state.known_path_map = _known_path_map
  state.details = {
    previous_snapshot_version: previous_snapshot_version,
    maps: {
      new:      new_path_map,
      deleted:  deleted_path_map,
      stable:   stable_path_map,
      unstable: unstable_path_map
    }
  }

  logger.debug { "Known Paths: #{_known_path_map.keys}" }
  logger.debug { "New Paths: #{new_path_map.map { |p, e| "#{p} #{e.stat}" }}" }

  logger.debug { "Deleted Paths: #{deleted_path_map.keys}" }
  logger.debug { "Unstable Paths: #{unstable_path_map.keys}" }
  logger.debug { "Stable Paths: #{stable_path_map.map { |p, e| [p, e[:stable_poll_count]] }}" }

  yield self if block_given?

end

#process_definition(watch_folder_def) ⇒ Object 



177
178
179
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 177

def process_definition(watch_folder_def)
  state.subscription = subscription_get_or_create(watch_folder_def)
end

#process_stable_entry(entry) ⇒ Object 



185
186
187
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 185

def process_stable_entry(entry)
  logger.debug { "Stable Entry Detected: #{entry.stat}" }
end

#subscription_get_or_create(watch_folder_def) ⇒ Object 



181
182
183
 
# File 'lib/envoi/aspera/watch_service/watch_folder.rb', line 181

def subscription_get_or_create(watch_folder_def)
  Subscription.get_or_create(client, watch_folder_def)
end