Sinatra Servlet That Forks Background Processes and Allows Polling Them Through DRb

gistfile1.rb

require 'sinatra'
require 'fileutils'
require 'cgi'
require 'drb'

TmpDir = './tmp/'
ServiceFile = 'services.txt'

def local_get(url)
  
  call(env.merge("PATH_INFO" => url)).last.join
  
end

def safe_write(path, content)
  
  if File.exist?(path)
    File.open(path).flock(File::LOCK_EX)
  end
  
  tmp = TmpDir + Random.rand(100000).to_s
  
  File.open(tmp, 'w+') do |f|
    f.write(content)
  end

  FileUtils.mv(tmp, path)
  
  File.open(path).flock(File::LOCK_UN)
  
end

get '/register/:job/:uri' do
  File.open(ServiceFile, 'a') do |file|
    file.write(
      params[:job] + " " +
      CGI.unescape(params[:uri]) + "\n"
    )
  end
  '0'
end

get '/process/:job' do

  job = params[:job]
  uri = nil
  pid = nil
  
  begin
    
    pid = fork do

      stack = []

      DRb.start_service nil, stack
      uri = CGI.escape(DRb.uri.to_s)
      url = "/register/#{job}/#{uri}"
      resp = local_get(url)

      unless resp.strip == '0'
        raise "Couldn't register job ID #{job}."
      end

      process(job, stack)

      stack << "#EOF#"

      DRb.thread.join

    end
    
    Process.detach(pid)
    
  rescue
    begin 
      `kill -9 #{pid}` unless pid
    rescue
      "Couldn't terminate child " +
      "process with PID #{pid}."
    end
  end

end

get '/poll/:job' do

  uri = nil
  service_line = nil

  lines = File.readlines(ServiceFile)
  
  lines.each do |line|
    job, server = line.split(' ')
    next unless job && server
    if job == params[:job]
      service_line = line
      uri = server
    end
  end

  item = nil
  
  begin
    DRb.start_service
    remote_array = DRbObject.new nil, uri
    item = remote_array.shift
  rescue
    raise "Couldn't start DRb service at #{uri}."
  end

  if item == '#EOF#'

    pruned = lines.delete_if do |line|
      line == service_line
    end
    
    safe_write(
      ServiceFile, 
      pruned.join("\n")
    )
    
    '#EOF#'
    
  else
    
    item.to_s
    
  end

end

def process(job, stack)
  # Do some processing.
  5.times do |i|
    stack << "job id #{job} - task #{i}"
  end
end