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 = 
5

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
 
# 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 = {}
end

Instance Method Details

#run ⇒ Object 



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

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