Fractor: Function-driven Ractors framework

require 'fractor'

# Define your work
class MyWork < Fractor::Work
  def initialize(value)
    super({ value: value })
  end

  def value
    input[:value]
  end
end

# Define your worker
class MyWorker < Fractor::Worker
  def process(work)
    result = work.value * 2
    Fractor::WorkResult.new(result: result, work: work)
  rescue => e
    Fractor::WorkResult.new(error: e.message, work: work)
  end
end

# Create supervisor and process work
supervisor = Fractor::Supervisor.new(
  worker_pools: [{ worker_class: MyWorker }]
)

supervisor.add_work_items([
  MyWork.new(1),
  MyWork.new(2),
  MyWork.new(3)
])

supervisor.run

puts "Results: #{supervisor.results.results.map(&:result)}"
# => Results: [2, 4, 6]

# 1. Global default (optional)
Fractor.configure do |config|
  config.default_worker_timeout = 60  # 60 seconds
end

# 2. Worker-level timeout
class FastWorker < Fractor::Worker
  timeout 10  # 10 second timeout for this worker

  def process(work)
    Fractor::WorkResult.new(result: work.input * 2, work: work)
  end
end

# 3. Per-work-item timeout (overrides worker timeout)
class MyWork < Fractor::Work
  def initialize(value, timeout: nil)
    super({ value: value }, timeout: timeout)
  end
end

supervisor = Fractor::Supervisor.new(
  worker_pools: [{ worker_class: FastWorker }]
)

# Mix work items with different timeouts
fast_work = MyWork.new(1, timeout: 5)      # 5 seconds
normal_work = MyWork.new(2)                 # uses worker's 10s timeout
slow_work = MyWork.new(3, timeout: 30)     # 30 seconds

supervisor.add_work_items([fast_work, normal_work, slow_work])
supervisor.run

# Create a persistent queue with automatic saving
queue = Fractor::PersistentWorkQueue.new("data/queue.json")

# Add work items - automatically saved
queue << MyWork.new(data1)
queue << MyWork.new(data2)

# Use with ContinuousServer for crash recovery
server = Fractor::ContinuousServer.new(
  worker_pools: [{ worker_class: MyWorker }],
  work_queue: queue
)

# Load any previous work items on startup
queue.load

server.run

# Create a cache with TTL (time-to-live)
cache = Fractor::ResultCache.new(ttl: 300)  # 5 minutes

# Or with size limit
cache = Fractor::ResultCache.new(max_size: 1000)

# Or with memory limit
cache = Fractor::ResultCache.new(max_memory: 1024 * 1024)  # 1MB

# Use the cache
result = cache.get(work) do
  # This block only runs if work is not cached
  expensive_operation(work)
end

# Check if work is cached
if cache.has?(work)
  puts "Result is cached!"
end

# Manual cache operations
cache.set(work, result)           # Store a result
cache.invalidate(work)            # Remove a cached result
cache.clear                       # Remove all cached results

# Get cache statistics
stats = cache.stats
puts "Cache hit rate: #{stats[:hit_rate]}%"

supervisor = Fractor::Supervisor.new(
  worker_pools: [{ worker_class: DataWorker }]
)

supervisor.add_work_items(dataset.map { |item| DataWork.new(item) })
supervisor.run

puts "Processed #{supervisor.results.results.size} items"

work_queue = Fractor::WorkQueue.new

server = Fractor::ContinuousServer.new(
  worker_pools: [{ worker_class: MessageWorker, num_workers: 4 }],
  work_queue: work_queue
)

server.on_result { |result| puts "Processed: #{result.result}" }
server.on_error { |error| puts "Error: #{error.error}" }

Thread.new { server.run }

# Add work dynamically
work_queue << MessageWork.new(client_id: 1, message: "Hello")

# Define workflow with simplified syntax
workflow = Fractor::Workflow.define("data-pipeline") do
  job :extract, ExtractWorker
  job :transform, TransformWorker, needs: :extract
  job :load, LoadWorker, needs: :transform
end

# Execute workflow
result = workflow.new.execute(input_data)