Merge remote branch 'origin/services-r10' into services-r11

Change-Id: Ib3828c66d6b572402e3bf0e0e70b1294f1f50d28

lib/services/api/clients/service_gateway_client.rb

@@ -1,5 +1,6 @@
 # Copyright (c) 2009-2011 VMware, Inc.
 require 'net/http'
+require 'uri'
 
 require 'services/api/const'
 require 'services/api/messages'
@@ -11,97 +12,166 @@ module Api
   end
 end
 
-class VCAP::Services::Api::ServiceGatewayClient
+module VCAP::Services::Api
+  class ServiceGatewayClient
+    METHODS_MAP = {
+      :get => Net::HTTP::Get,
+      :post=> Net::HTTP::Post,
+      :put => Net::HTTP::Put,
+      :delete => Net::HTTP::Delete,
+    }
 
-  class UnexpectedResponse < StandardError
-    attr_reader :response
+    # Public: Indicate gateway client encounter an unexpcted error,
+    # such as can't connect to gateway or can't decode response.
+    #
+    class UnexpectedResponse < StandardError; end
+
+    # Pubilc: Indicate an error response from gateway
+    #
+    class ErrorResponse < StandardError
+      attr_reader :status, :error
+
+      # status - the http status
+      # error  - a ServiceErrorResponse object
+      #
+      def initialize(status, error)
+        @status = status
+        @error = error
+      end
+
+      def to_s
+        "Reponse status:#{status},error:[#{error.extract}]"
+      end
+    end
 
-    def initialize(resp)
-      @response = resp
+    class NotFoundResponse < ErrorResponse
+      def initialize(error)
+        super(404, error)
+      end
     end
-  end
 
-  attr_reader :host, :port, :token
+    class GatewayInternalResponse < ErrorResponse
+      def initialize(error)
+        super(503, error)
+      end
+    end
 
-  def initialize(host, token, port=80)
-    @host  = host
-    @port  = port
-    @token = token
-    @hdrs  = {
-      'Content-Type' => 'application/json',
-      VCAP::Services::Api::GATEWAY_TOKEN_HEADER => @token
-    }
-  end
+    attr_reader :host, :port, :token
+    def initialize(url, token, timeout, opts={})
+      @url = url
+      @timeout = timeout
+      @token = token
+      @hdrs  = {
+        'Content-Type' => 'application/json',
+        GATEWAY_TOKEN_HEADER => @token
+      }
+    end
 
-  def provision(args)
-    msg = VCAP::Services::Api::GatewayProvisionRequest.new(args)
-    resp = perform_request(Net::HTTP::Post, '/gateway/v1/configurations', msg)
-    VCAP::Services::Api::GatewayProvisionResponse.decode(resp.body)
-  end
+    def provision(args)
+      msg = GatewayProvisionRequest.new(args)
+      resp = perform_request(:post, '/gateway/v1/configurations', msg)
+      GatewayProvisionResponse.decode(resp)
+    end
 
-  def unprovision(args)
-    perform_request(Net::HTTP::Delete, "/gateway/v1/configurations/#{args[:service_id]}")
-  end
+    def unprovision(args)
+      resp = perform_request(:delete, "/gateway/v1/configurations/#{args[:service_id]}")
+      EMPTY_REQUEST
+    end
 
-  def create_snapshot(args)
-    resp = perform_request(Net::HTTP::Post, "/gateway/v1/configurations/#{args[:service_id]}/snapshots")
-    VCAP::Services::Api::Job.decode(resp.body)
-  end
+    def create_snapshot(args)
+      resp = perform_request(:post, "/gateway/v1/configurations/#{args[:service_id]}/snapshots")
+      Job.decode(resp)
+    end
 
-  def enum_snapshots(args)
-    resp = perform_request(Net::HTTP::Get, "/gateway/v1/configurations/#{args[:service_id]}/snapshots")
-    VCAP::Services::Api::SnapshotList.decode(resp.body)
-  end
+    def enum_snapshots(args)
+      resp = perform_request(:get, "/gateway/v1/configurations/#{args[:service_id]}/snapshots")
+      SnapshotList.decode(resp)
+    end
 
-  def snapshot_details(args)
-    resp = perform_request(Net::HTTP::Get, "/gateway/v1/configurations/#{args[:service_id]}/snapshots/#{args[:snapshot_id]}")
-    VCAP::Services::Api::Snapshot.decode(resp.body)
-  end
+    def snapshot_details(args)
+      resp = perform_request(:get, "/gateway/v1/configurations/#{args[:service_id]}/snapshots/#{args[:snapshot_id]}")
+      Snapshot.decode(resp)
+    end
 
-  def rollback_snapshot(args)
-    resp = perform_request(Net::HTTP::Put, "/gateway/v1/configurations/#{args[:service_id]}/snapshots/#{args[:snapshot_id]}")
-    VCAP::Services::Api::Job.decode(resp.body)
-  end
+    def rollback_snapshot(args)
+      resp = perform_request(:put, "/gateway/v1/configurations/#{args[:service_id]}/snapshots/#{args[:snapshot_id]}")
+      Job.decode(resp)
+    end
 
-  def serialized_url(args)
-    resp = perform_request(Net::HTTP::Get, "/gateway/v1/configurations/#{args[:service_id]}/serialized/url")
-    VCAP::Services::Api::Job.decode(resp.body)
-  end
+    def delete_snapshot(args)
+      resp = perform_request(:delete, "/gateway/v1/configurations/#{args[:service_id]}/snapshots/#{args[:snapshot_id]}")
+      Job.decode(resp)
+    end
 
-  def import_from_url(args)
-    resp = perform_request(Net::HTTP::Put, "/gateway/v1/configurations/#{args[:service_id]}/serialized/url", args[:msg])
-    VCAP::Services::Api::Job.decode(resp.body)
-  end
+    def serialized_url(args)
+      resp = perform_request(:get, "/gateway/v1/configurations/#{args[:service_id]}/serialized/url")
+      Job.decode(resp)
+    end
 
-  def import_from_data(args)
-    resp = perform_request(Net::HTTP::Put, "/gateway/v1/configurations/#{args[:service_id]}/serialized/data", args[:msg])
-    VCAP::Services::Api::Job.decode(resp.body)
-  end
+    def import_from_url(args)
+      resp = perform_request(:put, "/gateway/v1/configurations/#{args[:service_id]}/serialized/url", args[:msg])
+      Job.decode(resp)
+    end
 
-  def job_info(args)
-    resp = perform_request(Net::HTTP::Get, "/gateway/v1/configurations/#{args[:service_id]}/jobs/#{args[:job_id]}")
-    VCAP::Services::Api::Job.decode(resp.body)
-  end
+    def import_from_data(args)
+      resp = perform_request(:put, "/gateway/v1/configurations/#{args[:service_id]}/serialized/data", args[:msg])
+      Job.decode(resp)
+    end
 
-  def bind(args)
-    msg = VCAP::Services::Api::GatewayBindRequest.new(args)
-    resp = perform_request(Net::HTTP::Post, "/gateway/v1/configurations/#{msg.service_id}/handles", msg)
-    VCAP::Services::Api::GatewayBindResponse.decode(resp.body)
-  end
+    def job_info(args)
+      resp = perform_request(:get, "/gateway/v1/configurations/#{args[:service_id]}/jobs/#{args[:job_id]}")
+      Job.decode(resp)
+    end
 
-  def unbind(args)
-    msg = VCAP::Services::Api::GatewayUnbindRequest.new(args)
-    perform_request(Net::HTTP::Delete, "/gateway/v1/configurations/#{msg.service_id}/handles/#{msg.handle_id}", msg)
-  end
+    def bind(args)
+      msg = GatewayBindRequest.new(args)
+      resp = perform_request(:post, "/gateway/v1/configurations/#{msg.service_id}/handles", msg)
+      GatewayBindResponse.decode(resp)
+    end
 
-  protected
+    def unbind(args)
+      msg = GatewayUnbindRequest.new(args)
+      perform_request(:delete, "/gateway/v1/configurations/#{msg.service_id}/handles/#{msg.handle_id}", msg)
+      EMPTY_REQUEST
+    end
 
-  def perform_request(klass, path, msg=VCAP::Services::Api::EMPTY_REQUEST)
-    req = klass.new(path, initheader=@hdrs)
-    req.body = msg.encode
-    resp = Net::HTTP.new(@host, @port).start {|http| http.request(req)}
-    raise UnexpectedResponse, resp unless resp.is_a? Net::HTTPOK
-    resp
+    protected
+
+    def perform_request(http_method, path, msg=VCAP::Services::Api::EMPTY_REQUEST)
+      result = nil
+      uri = URI.parse(@url)
+      if EM.reactor_running?
+        url = uri.merge!(path)
+        http = AsyncHttpRequest.fibered(url, @token, http_method, @timeout, msg)
+        raise UnexpectedResponse, "Error sending request #{msg.extract.to_json} to gateway #{@url}: #{http.error}" unless http.error.empty?
+        code = http.response_header.status.to_i
+        body = http.response
+      else
+        klass = METHODS_MAP[http_method]
+        req = klass.new(path, initheader=@hdrs)
+        req.body = msg.encode
+        resp = Net::HTTP.new(uri.host, uri.port).start {|http| http.request(req)}
+        code = resp.code.to_i
+        body = resp.body
+      end
+      case code
+      when 200
+        body
+      when 404
+        err = ServiceErrorResponse.decode(body)
+        raise NotFoundResponse.new(err)
+      when 503
+        err = ServiceErrorResponse.decode(body)
+        raise GatewayInternalResponse.new(err)
+      else
+        begin
+          # try to decode the response
+          err = ServiceErrorResponse.decode(body)
+          raise ErrorResponse.new(code, err)
+        rescue => e
+          raise UnexpectedResponse, "Can't decode gateway response. status code:#{code}, response body:#{body}"
+        end
+      end
+    end
   end
-
 end

lib/services/api/messages.rb

@@ -148,6 +148,11 @@ class SerializedURL < JsonMessage
       class SerializedData < JsonMessage
         required :data,  String
       end
+
+      class ServiceErrorResponse < JsonMessage
+        required :code, Integer
+        required :description, String
+      end
     end
   end
 end

spec/spec_helper.rb

@@ -13,6 +13,8 @@
 require "vcap/config"
 require "vcap/priority_queue"
 require 'vcap/quota'
+require 'services/api/clients/service_gateway_client'
+require 'services/api/async_requests'
 require 'benchmark'
 
 RSpec::Matchers.define :take_less_than do |n|

spec/unit/service_gateway_client_spec.rb

@@ -0,0 +1,81 @@
+# Copyright (c) 2009-2012 VMware, Inc.
+require 'spec_helper'
+
+module VCAP::Services::Api
+  class ServiceGatewayClient
+    public :perform_request
+  end
+end
+describe VCAP::Services::Api::ServiceGatewayClient do
+  describe '#perform_request' do
+    before :all do
+      @url = "http://localhost"
+      @token = "mytoken"
+      @timeout = 10
+    end
+
+    it "should use async http client when EM is running" do
+      client = VCAP::Services::Api::ServiceGatewayClient.new(@url, @token, @timeout)
+      EM.should_receive(:reactor_running?).and_return true
+
+      path = "/path1"
+      resp = mock("resq")
+      message = "data"
+      resp.should_receive(:response).and_return(message)
+      resp.should_receive(:error).and_return []
+
+      resp_header = mock("resq_header")
+      resp_header.should_receive(:status).and_return(200)
+      resp.should_receive(:response_header).and_return resp_header
+      http_method = :get
+
+      VCAP::Services::Api::AsyncHttpRequest.should_receive(:fibered).with(anything, @token, http_method, @timeout, anything).and_return resp
+
+      result = client.perform_request(http_method, path)
+      result.should == message
+    end
+
+    it "should use net/http client when EM is not running" do
+      client = VCAP::Services::Api::ServiceGatewayClient.new(@url, @token, @timeout)
+      EM.should_receive(:reactor_running?).and_return nil
+
+      path = "/path1"
+      resp = mock("resq")
+      message = "data"
+      resp.should_receive(:body).and_return(message)
+      resp.should_receive(:code).and_return 200
+      resp.should_receive(:start).and_return resp
+
+      http_method = :get
+
+      Net::HTTP.should_receive(:new).with("localhost", 80).and_return resp
+
+      result = client.perform_request(http_method, path)
+      result.should == message
+    end
+
+
+    it "should should raise error with none 200 response" do
+      client = VCAP::Services::Api::ServiceGatewayClient.new(@url, @token, @timeout)
+      EM.should_receive(:reactor_running?).any_number_of_times.and_return nil
+
+      path = "/path1"
+      resp = mock("resq")
+      resp.should_receive(:body).and_return(
+        {:code => 40400, :description=> "not found"}.to_json,
+        {:code => 50300, :description=> "internal"}.to_json,
+        {:code => 50100, :description=> "not done yet"}.to_json,
+      )
+      resp.should_receive(:code).and_return(404, 503, 500)
+      resp.should_receive(:start).any_number_of_times.and_return resp
+
+      http_method = :get
+
+      Net::HTTP.should_receive(:new).with("localhost", 80).any_number_of_times.and_return resp
+
+      expect {client.perform_request(http_method, path)}.should raise_error(VCAP::Services::Api::ServiceGatewayClient::NotFoundResponse)
+      expect {client.perform_request(http_method, path)}.should raise_error(VCAP::Services::Api::ServiceGatewayClient::GatewayInternalResponse)
+      expect {client.perform_request(http_method, path)}.should raise_error(VCAP::Services::Api::ServiceGatewayClient::UnexpectedResponse)
+    end
+  end
+end