lib/upr/json.rb

   1 # -*- encoding: binary -*-
   2 begin
   3   require 'json'
   4 rescue LoadError
   5   raise LoadError, "either json or json-pure is required"
   6 end
   7 require 'rack'
   8
   9 module Upr
  10
  11   # JSON protocol based on Lighttpd's mod_uploadprogress
  12   # http://trac.lighttpd.net/trac/wiki/Docs:ModUploadProgress
  13   class JSON < Struct.new(:frequency, :backend, :upload_id)
  14
  15     include Params
  16
  17     # We use this in case length is nil when clients send chunked uploads
  18     INT_MAX = 0x7fffffff
  19
  20     SLEEP_CLASS = defined?(Actor) ? Actor : Kernel
  21
  22     RESP_HEADERS = {
  23       'Content-Type' => 'application/json',
  24       'Cache-Control' => 'no-cache',
  25     }
  26
  27     def initialize(options = {})
  28       super(options[:frequency] || 1, options[:backend], options[:upload_id])
  29
  30       # support :drb for compatibility with mongrel_upload_progress
  31       if options[:drb]
  32         backend and raise ArgumentError, ":backend and :drb are incompatible"
  33         require 'drb'
  34         DRb.start_service
  35         self.backend = DRbObject.new(nil, options[:drb])
  36       elsif String === backend
  37         # allow people to use strings in case their backend gets
  38         # lazy-loaded (like an ActiveRecord model)
  39         self.backend = eval(backend)
  40       elsif backend.nil?
  41         raise ArgumentError, "backend MUST be specified"
  42       end
  43
  44       # only for use with rails_proc
  45       upload_id.nil? and self.upload_id = options[:env]
  46     end
  47
  48     def rails_render_options
  49       env = upload_id
  50       self.upload_id = extract_upload_id(env)
  51       text = if Rack::Request.new(env).GET.include?("long")
  52         Proc.new { |response,output| each { |line| output.write(line) } }
  53       else
  54         _once
  55       end
  56       { :content_type => 'application/json', :text => text }
  57     end
  58
  59     def _once
  60       if status = backend.read(upload_id)
  61         if status.done?
  62           _json_object(:state => 'done')
  63         elsif status.seen == 0
  64           _json_object(:state => 'starting')
  65         elsif status.error?
  66           _error_msg("upload failed")
  67         else
  68           _update_msg(status)
  69         end
  70       else
  71         timeout = Time.now + 2
  72         until status = backend.read(upload_id)
  73           SLEEP_CLASS.sleep(0.1)
  74           return _error_msg("couldn't get status") if Time.now > timeout
  75         end
  76         _json_object(:state => 'starting')
  77       end
  78     end
  79
  80     # Rack interface reservced for future use with streaming AJAX
  81     def call(env)
  82       if uid = extract_upload_id(env)
  83         _wrap(env, uid)
  84       else
  85         [ 400, RESP_HEADERS.dup, [ _error_msg("upload_id not given") ] ]
  86       end
  87     end
  88
  89     # Rack interface reservced for future use with streaming AJAX
  90     def each(&block)
  91       sleeper = defined?(Actor) ? Actor : Kernel
  92       timeout = Time.now + 2
  93       eol = ";\n"
  94       yield _json_object(:state => 'starting') << eol
  95       begin
  96         until status = backend.read(upload_id)
  97           sleeper.sleep(0.1)
  98           break if Time.now > timeout
  99         end
 100         if status
 101           begin
 102             yield _update_msg(status) << eol
 103             break if status.done?
 104             sleeper.sleep(frequency)
 105           end while status = backend.read(upload_id)
 106           yield _json_object(:state => 'done') << eol
 107         else
 108           yield _error_msg("couldn't get status") << eol
 109         end
 110       rescue => e
 111         yield _error_msg(e.message) << eol
 112       end
 113     end
 114
 115     # Rack interface reservced for future use with streaming AJAX
 116     def _wrap(env, uid)
 117       _self = dup
 118       _self.upload_id = uid
 119       [ 200, RESP_HEADERS.dup, _self ]
 120     end
 121
 122     def _error_msg(msg)
 123       _json_object(:state => 'error', :status => 400, :message => msg)
 124     end
 125
 126     def _json_object(options)
 127       # $stderr.syswrite "#{options.inspect} #{$upr.inspect}\n"
 128       options.to_json
 129     end
 130
 131     def _update_msg(status)
 132       raise "client error" if status.error?
 133       received = status.seen
 134       size = status.length || INT_MAX
 135       _json_object(:state => 'uploading', :size => size, :received => received)
 136     end
 137
 138   end
 139 end