Class: CouchProxy::Rack::Base

Inherits:
Object
  • Object
show all
Defined in:
lib/couchproxy/rack/base.rb

Constant Summary collapse

APPLICATION_JSON = 
"application/json".freeze
TEXT_PLAIN = 
"text/plain;charset=utf-8".freeze
DESIGN_ID = 
/^_design\/.+/
METHODS = 
[:get, :put, :post, :delete, :head].freeze
INVALID_JSON = 
'{"error":"bad_request","reason":"invalid UTF-8 JSON"}'.freeze
SERVER_VERSION = 
"CouchProxy/#{CouchProxy::VERSION}".freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(request, cluster) ⇒ Base

Returns a new instance of Base.



15
16
17
 
# File 'lib/couchproxy/rack/base.rb', line 15

def initialize(request, cluster)
  @request, @cluster = request, cluster
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(name) ⇒ Object 



19
20
21
22
23
24
 
# File 'lib/couchproxy/rack/base.rb', line 19

def method_missing(name)
  allowed = (methods & METHODS).map {|m| m.to_s.upcase }.sort.join(',')
  body = "{\"error\:\"method_not_allowed\",\"reason\":\"Only #{allowed} allowed\"}"
  headers = response_headers.tap {|h| h['Allow'] = allowed }
  send_response(405, headers, [body])
end

Instance Attribute Details

#cluster ⇒ Object (readonly)

Returns the value of attribute cluster.



13
14
15
 
# File 'lib/couchproxy/rack/base.rb', line 13

def cluster
  @cluster
end

#request ⇒ Object (readonly)

Returns the value of attribute request.



13
14
15
 
# File 'lib/couchproxy/rack/base.rb', line 13

def request
  @request
end

Instance Method Details

#proxy_to(node, &finish) ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/couchproxy/rack/base.rb', line 26

def proxy_to(node, &finish)
  head_proxy_to(node, &finish) if @request.head?

  body, started = DeferrableBody.new, false
  uri = "#{node.uri}#{@request.fullpath}"
  http = EM::HttpRequest.new(uri)
  res = http.send(@request.request_method.downcase,
    :head => proxy_headers, :body => @request.content)
  res.stream do |chunk|
    unless started
      started = true
      head = normalize(res.response_header).tap do |h|
        h['Server'] = SERVER_VERSION
        if res.response_header.location
          h['Location'] = rewrite_location(res.response_header.location)
        end
      end
      send_response(res.response_header.status, head, body)
    end
    body.call([chunk])
  end
  res.callback do
    body.succeed
    finish.call if finish
  end
  res.errback { send_error_response }
end

#proxy_to_all_nodes(&callback) ⇒ Object 



72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/couchproxy/rack/base.rb', line 72

def proxy_to_all_nodes(&callback)
  method = request.request_method.downcase
  multi = EM::MultiRequest.new
  @cluster.nodes.each do |n|
    uri = "#{n.uri}#{@request.fullpath}"
    req = EM::HttpRequest.new(uri).send(method,
      :head => proxy_headers, :body => @request.content)
    multi.add(req)
  end
  multi.callback(&multi(&callback)) if callback
end

#proxy_to_all_partitions(&callback) ⇒ Object 



84
85
86
87
88
89
90
91
92
93
 
# File 'lib/couchproxy/rack/base.rb', line 84

def proxy_to_all_partitions(&callback)
  method = request.request_method.downcase
  multi = EM::MultiRequest.new
  @cluster.partitions.each do |p|
    uri = "#{p.node.uri}#{@request.rewrite_proxy_url(p.num)}#{query_string}"
    multi.add EM::HttpRequest.new(uri).send(method,
      :head => proxy_headers, :body => @request.content)
  end
  multi.callback(&multi(&callback)) if callback
end

#proxy_to_any_node ⇒ Object 



62
63
64
 
# File 'lib/couchproxy/rack/base.rb', line 62

def proxy_to_any_node
  proxy_to(@cluster.any_node)
end

#proxy_to_any_partition ⇒ Object 



66
67
68
69
70
 
# File 'lib/couchproxy/rack/base.rb', line 66

def proxy_to_any_partition
  partition = @cluster.any_partition
  request.rewrite_proxy_url!(partition.num)
  proxy_to(partition.node)
end

#replicate_to_all_partitions(source, *doc_ids, &callback) ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/couchproxy/rack/base.rb', line 95

def replicate_to_all_partitions(source, *doc_ids, &callback)
  multi = EM::MultiRequest.new
  (@cluster.partitions - [source]).each do |p|
    task = {
      :source => source.uri(request.db_name),
      :target => p.uri(request.db_name),
      :doc_ids => doc_ids.flatten}
    multi.add EM::HttpRequest.new("#{p.node.uri}/_replicate").post(
      :head => proxy_headers, :body => task.to_json)
  end
  multi.callback(&multi(&callback)) if callback
end

#rewrite_location(uri) ⇒ Object 



54
55
56
57
58
59
60
 
# File 'lib/couchproxy/rack/base.rb', line 54

def rewrite_location(uri)
  URI.parse(request.url).tap do |req|
    req.query, req.user, req.password = nil
    req.path = URI.parse(uri).path.gsub(
      /^\/#{request.db_name}_\d+/, "/#{request.db_name}")
  end.to_s
end

#send_error_response ⇒ Object 



125
126
127
 
# File 'lib/couchproxy/rack/base.rb', line 125

def send_error_response
  send_response(503, response_headers, [])
end

#send_response(*args) ⇒ Object 



121
122
123
 
# File 'lib/couchproxy/rack/base.rb', line 121

def send_response(*args)
  @request.env['async.callback'].call(args)
end

#uuids(count, &callback) ⇒ Object 



108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/couchproxy/rack/base.rb', line 108

def uuids(count, &callback)
  http = EM::HttpRequest.new("#{@cluster.any_node.uri}/_uuids?count=#{count}").get
  http.errback { callback.call(nil) }
  http.callback do |res|
    if res.response_header.status == 200
      uuids = JSON.parse(res.response)['uuids']
      callback.call(uuids)
    else
      callback.call(nil)
    end
  end
end