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     # our default response headers, we need to set no-transform to
  23     # prevent deflaters from compressing our already-small small input
  24     # and also to prevent buffering/corking of the response inside
  25     # deflater buffers.
  26     RESPONSE_HEADERS = {
  27       'Content-Type' => 'application/json',
  28       'Cache-Control' => 'no-cache, no-transform',
  29     }
  30
  31     def initialize(options = {})
  32       super(options[:frequency] || 1, options[:backend], options[:upload_id])
  33
  34       # support :drb for compatibility with mongrel_upload_progress
  35       if options[:drb]
  36         backend and raise ArgumentError, ":backend and :drb are incompatible"
  37         require 'drb'
  38         DRb.start_service
  39         self.backend = DRbObject.new(nil, options[:drb])
  40       elsif String === backend
  41         # allow people to use strings in case their backend gets
  42         # lazy-loaded (like an ActiveRecord model)
  43         self.backend = eval(backend)
  44       elsif backend.nil?
  45         raise ArgumentError, "backend MUST be specified"
  46       end
  47
  48       # only for use with rails_proc
  49       upload_id.nil? and self.upload_id = options[:env]
  50     end
  51
  52     def rails_render_options
  53       env = upload_id
  54       self.upload_id = extract_upload_id(env)
  55       text = if Rack::Request.new(env).GET.include?("long")
  56         Proc.new { |response,output| each { |line| output.write(line) } }
  57       else
  58         _once
  59       end
  60       { :content_type => 'application/json', :text => text }
  61     end
  62
  63     def _once
  64       if status = backend.read(upload_id)
  65         if status.done?
  66           _json_object(:state => 'done')
  67         elsif status.seen == 0
  68           _json_object(:state => 'starting')
  69         elsif status.error?
  70           _error_msg("upload failed")
  71         else
  72           _update_msg(status)
  73         end
  74       else
  75         timeout = Time.now + 2
  76         until status = backend.read(upload_id)
  77           SLEEP_CLASS.sleep(0.1)
  78           return _error_msg("couldn't get status") if Time.now > timeout
  79         end
  80         _json_object(:state => 'starting')
  81       end
  82     end
  83
  84     # Rack interface reservced for future use with streaming AJAX
  85     def call(env)
  86       if uid = extract_upload_id(env)
  87         _wrap(env, uid)
  88       else
  89         [ 400, RESPONSE_HEADERS.dup, [ _error_msg("upload_id not given") ] ]
  90       end
  91     end
  92
  93     # Rack interface reservced for future use with streaming AJAX
  94     def each(&block)
  95       sleeper = defined?(Actor) ? Actor : Kernel
  96       timeout = Time.now + 2
  97       eol = ";\n"
  98       yield _json_object(:state => 'starting') << eol
  99       begin
 100         until status = backend.read(upload_id)
 101           sleeper.sleep(0.1)
 102           break if Time.now > timeout
 103         end
 104         if status
 105           begin
 106             yield _update_msg(status) << eol
 107             break if status.done?
 108             sleeper.sleep(frequency)
 109           end while status = backend.read(upload_id)
 110           yield _json_object(:state => 'done') << eol
 111         else
 112           yield _error_msg("couldn't get status") << eol
 113         end
 114       rescue => e
 115         yield _error_msg(e.message) << eol
 116       end
 117     end
 118
 119     # Rack interface reservced for future use with streaming AJAX
 120     def _wrap(env, uid)
 121       _self = dup
 122       _self.upload_id = uid
 123       [ 200, RESPONSE_HEADERS.dup, _self ]
 124     end
 125
 126     def _error_msg(msg)
 127       _json_object(:state => 'error', :status => 400, :message => msg)
 128     end
 129
 130     def _json_object(options)
 131       # $stderr.syswrite "#{options.inspect} #{$upr.inspect}\n"
 132       options.to_json
 133     end
 134
 135     def _update_msg(status)
 136       raise "client error" if status.error?
 137       received = status.seen
 138       size = status.length || INT_MAX
 139       _json_object(:state => 'uploading', :size => size, :received => received)
 140     end
 141
 142   end
 143 end