Class: GitFit::Sync::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/git_fit/sync/runner.rb

Constant Summary collapse

POISON = 
Object.new.freeze
BATCH_SIZE = 
10

Instance Method Summary collapse

Constructor Details

#initialize(sources:, db:, config:, activity_filter: nil, privacy: nil, num_workers: 3, time_budget: nil, total_timeout: nil) ⇒ Runner

Returns a new instance of Runner.



11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/git_fit/sync/runner.rb', line 11

def initialize(sources:, db:, config:, activity_filter: nil,
               privacy: nil, num_workers: 3, time_budget: nil, total_timeout: nil)
  @sources = sources
  @db = db
  @config = config
  @activity_filter = activity_filter
  @privacy = privacy
  @num_workers = [num_workers.to_i, 1].max
  @time_budget = time_budget.is_a?(Numeric) ? time_budget.to_f : nil
  @total_timeout = total_timeout.is_a?(Numeric) ? total_timeout.to_f : nil
  @sync_start = Time.now
  @watchdog = nil

  @pending_ids = {}
  @adapters = {}
  @source_queue = Queue.new
  @output_queue = Queue.new
  @stats_lock = Mutex.new
  @unfinished = 0
  @completed = {}
  @timed_out = false
  @shutdown = false
  @source_start_times = {}
  @source_activated = {}
  @run_totals = {}
  @run_processed = {}
end

Instance Method Details

#run ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/git_fit/sync/runner.rb', line 39

def run
  output_thread = Thread.new { output_loop }
  start_watchdog if @total_timeout
  init_pending
  unless @source_queue.empty?
    worker_threads = @num_workers.times.map { Thread.new { worker_loop } }
    worker_threads.each(&:join)
  end
  output_queue << POISON
  output_thread.join(60) || output_thread.kill

  # TODO: Dedup 需要独立设计三种语义后再实现
  #   A. 同源防重: upsert_activity 已通过 run_id 唯一约束天然保证
  #   B. 跨设备同活动: 需轨迹相似度匹配 + 时间窗口聚类
  #   C. 跨平台迁移复制: 需 external_id 映射表 + 数据指纹比对
  #   触发时机、匹配算法、合并策略见 docs/dedup-design.md
  # Dedup::Service.new(@db).run if defined?(Dedup::Service) && ...
ensure
  @watchdog&.kill
end