lib/ably/rest/client.rb in ably-1.0.7 vs lib/ably/rest/client.rb in ably-1.1.0
- old
+ new
@@ -28,10 +28,12 @@
request_timeout: 10,
max_retry_duration: 15,
max_retry_count: 3
}.freeze
+ FALLBACK_RETRY_TIMEOUT = 10 * 60
+
def_delegators :auth, :client_id, :auth_options
# Custom environment to use such as 'sandbox' when testing the client library against an alternate Ably environment
# @return [String]
attr_reader :environment
@@ -81,14 +83,27 @@
# The list of fallback hosts to be used by this client
# if empty or nil then fallback host functionality is disabled
attr_reader :fallback_hosts
- # Whethere the {Client} has to add a random identifier to the path of a request
+ # Whether the {Client} has to add a random identifier to the path of a request
# @return [Boolean]
attr_reader :add_request_ids
+ # Retries are logged by default to warn and error. When true, retries are logged at info level
+ # @return [Boolean]
+ # @api private
+ attr_reader :log_retries_as_info
+
+ # True when idempotent publishing is enabled for all messages published via REST.
+ # When this feature is enabled, the client library will add a unique ID to every published message (without an ID)
+ # ensuring any failed published attempts (due to failures such as HTTP requests failing mid-flight) that are
+ # automatically retried will not result in duplicate messages being published to the Ably platform.
+ # Note: This is a beta unsupported feature!
+ # @return [Boolean]
+ attr_reader :idempotent_rest_publishing
+
# Creates a {Ably::Rest::Client Rest Client} and configures the {Ably::Auth} object for the connection.
#
# @param [Hash,String] options an options Hash used to configure the client and the authentication, or String with an API key or Token ID
# @option options [Boolean] :tls (true) When false, TLS is disabled. Please note Basic Auth is disallowed without TLS as secrets cannot be transmitted over unsecured connections.
# @option options [String] :key API key comprising the key name and key secret in a single string
@@ -115,11 +130,14 @@
# @option options [Integer] :http_max_retry_count (3) maximum number of fallback host retries for HTTP requests that fail due to network issues or server problems
# @option options [Integer] :http_max_retry_duration (15 seconds) maximum elapsed time in which fallback host retries for HTTP requests will be attempted i.e. if the first default host attempt takes 5s, and then the subsequent fallback retry attempt takes 7s, no further fallback host attempts will be made as the total elapsed time of 12s exceeds the default 10s limit
#
# @option options [Boolean] :fallback_hosts_use_default (false) When true, forces the user of fallback hosts even if a non-default production endpoint is being used
# @option options [Array<String>] :fallback_hosts When an array of fallback hosts are provided, these fallback hosts are always used if a request fails to the primary endpoint. If an empty array is provided, the fallback host functionality is disabled
+ # @option options [Integer] :fallback_retry_timeout (600 seconds) amount of time in seconds a REST client will continue to use a working fallback host when the primary fallback host has previously failed
+ #
# @option options [Boolean] :add_request_ids (false) When true, adds a unique request_id to each request sent to Ably servers. This is handy when reporting issues, because you can refer to a specific request.
+ # @option options [Boolean] :idempotent_rest_publishing (false if ver < 1.2) When true, idempotent publishing is enabled for all messages published via REST
#
# @return [Ably::Rest::Client]
#
# @example
# # create a new client authenticating with basic auth
@@ -138,23 +156,26 @@
else
{ token: options }
end
end
- @realtime_client = options.delete(:realtime_client)
- @tls = options.delete(:tls) == false ? false : true
- @environment = options.delete(:environment) # nil is production
- @environment = nil if [:production, 'production'].include?(@environment)
- @protocol = options.delete(:protocol) || :msgpack
- @debug_http = options.delete(:debug_http)
- @log_level = options.delete(:log_level) || ::Logger::WARN
- @custom_logger = options.delete(:logger)
- @custom_host = options.delete(:rest_host)
- @custom_port = options.delete(:port)
- @custom_tls_port = options.delete(:tls_port)
- @add_request_ids = options.delete(:add_request_ids)
+ @realtime_client = options.delete(:realtime_client)
+ @tls = options.delete(:tls) == false ? false : true
+ @environment = options.delete(:environment) # nil is production
+ @environment = nil if [:production, 'production'].include?(@environment)
+ @protocol = options.delete(:protocol) || :msgpack
+ @debug_http = options.delete(:debug_http)
+ @log_level = options.delete(:log_level) || ::Logger::WARN
+ @custom_logger = options.delete(:logger)
+ @custom_host = options.delete(:rest_host)
+ @custom_port = options.delete(:port)
+ @custom_tls_port = options.delete(:tls_port)
+ @add_request_ids = options.delete(:add_request_ids)
+ @log_retries_as_info = options.delete(:log_retries_as_info)
+ @idempotent_rest_publishing = options.delete(:idempotent_rest_publishing) || Ably.major_minor_version_numeric > 1.1
+
if options[:fallback_hosts_use_default] && options[:fallback_jhosts]
raise ArgumentError, "fallback_hosts_use_default cannot be set to trye when fallback_jhosts is also provided"
end
@fallback_hosts = case
when options.delete(:fallback_hosts_use_default)
@@ -165,10 +186,14 @@
[]
else
Ably::FALLBACK_HOSTS
end
+ options[:fallback_retry_timeout] ||= FALLBACK_RETRY_TIMEOUT
+
+ # Take option keys prefixed with `http_`, remove the http_ and
+ # check if the option exists in HTTP_DEFAULTS. If so, update http_defaults
@http_defaults = HTTP_DEFAULTS.dup
options.each do |key, val|
if http_key = key[/^http_(.+)/, 1]
@http_defaults[http_key.to_sym] = val if val && @http_defaults.has_key?(http_key.to_sym)
end
@@ -189,12 +214,16 @@
end
end
raise ArgumentError, 'Protocol is invalid. Must be either :msgpack or :json' unless [:msgpack, :json].include?(@protocol)
token_params = options.delete(:default_token_params) || {}
- @options = options
- @auth = Auth.new(self, token_params, options)
+ @options = options
+ init_auth_options = options.select do |key, _|
+ Auth::AUTH_OPTIONS_KEYS.include?(key.to_s)
+ end
+
+ @auth = Auth.new(self, token_params, init_auth_options)
@channels = Ably::Rest::Channels.new(self)
@encoders = []
options.freeze
@@ -272,10 +301,28 @@
# @api private
def post(path, params, options = {})
raw_request(:post, path, params, options)
end
+ # Perform an HTTP PUT request to the API using configured authentication
+ #
+ # @return [Faraday::Response]
+ #
+ # @api private
+ def put(path, params, options = {})
+ raw_request(:put, path, params, options)
+ end
+
+ # Perform an HTTP DELETE request to the API using configured authentication
+ #
+ # @return [Faraday::Response]
+ #
+ # @api private
+ def delete(path, params, options = {})
+ raw_request(:delete, path, params, options)
+ end
+
# Perform an HTTP request to the Ably API
# This is a convenience for customers who wish to use bleeding edge REST API functionality
# that is either not documented or is not included in the API for our client libraries.
# The REST client library provides a function to issue HTTP requests to the Ably endpoints
# with all the built in functionality of the library such as authentication, paging,
@@ -291,18 +338,18 @@
def request(method, path, params = {}, body = nil, headers = {}, options = {})
raise "Method #{method.to_s.upcase} not supported" unless [:get, :put, :post].include?(method.to_sym)
response = case method.to_sym
when :get
- reauthorize_on_authorisation_failure do
+ reauthorize_on_authorization_failure do
send_request(method, path, params, headers: headers)
end
when :post
path_with_params = Addressable::URI.new
path_with_params.query_values = params || {}
query = path_with_params.query
- reauthorize_on_authorisation_failure do
+ reauthorize_on_authorization_failure do
send_request(method, "#{path}#{"?#{query}" unless query.nil? || query.empty?}", body, headers: headers)
end
end
paginated_options = {
@@ -320,10 +367,24 @@
rescue Exceptions::InvalidRequest, Exceptions::ServerError => e
response = Models::HttpPaginatedResponse::ErrorResponse.new(e.status, e.code, e.message)
Models::HttpPaginatedResponse.new(response, path, self)
end
+ # The local device detilas
+ # @return [Ably::Models::LocalDevice]
+ #
+ # @note This is unsupported in the Ruby library
+ def device
+ raise Ably::Exceptions::PushNotificationsNotSupported, 'This device does not support receiving or subscribing to push notifications. The local device object is not unavailable'
+ end
+
+ # Push notification object for publishing and managing push notifications
+ # @return [Ably::Rest::Push]
+ def push
+ @push ||= Push.new(self)
+ end
+
# @!attribute [r] endpoint
# @return [URI::Generic] Default Ably REST endpoint used for all requests
def endpoint
endpoint_for_host(custom_host || [@environment, DOMAIN].compact.join('-'))
end
@@ -388,11 +449,10 @@
#
# @api private
def fallback_connection
unless defined?(@fallback_connections) && @fallback_connections
@fallback_connections = fallback_hosts.shuffle.map { |host| Faraday.new(endpoint_for_host(host).to_s, connection_options) }
- @fallback_connections << Faraday.new(endpoint.to_s, connection_options) # Try the original host last if all fallbacks have been used
end
@fallback_index ||= 0
@fallback_connections[@fallback_index % @fallback_connections.count].tap do
@fallback_index += 1
@@ -419,17 +479,50 @@
else
http_defaults.fetch(:request_timeout)
end
end
+ # If the primary host endpoint fails, and a subsequent fallback host succeeds, the fallback
+ # host that succeeded is used for +ClientOption+ +fallback_retry_timeout+ seconds to avoid
+ # retries to known failing hosts for a short period of time.
+ # See https://github.com/ably/docs/pull/554, spec id #RSC15f
+ #
+ # @return [nil, String] Returns nil (falsey) if the primary host is being used, or the currently used host if a fallback host is currently preferred
+ def using_preferred_fallback_host?
+ if preferred_fallback_connection && (preferred_fallback_connection.fetch(:expires_at) > Time.now)
+ preferred_fallback_connection.fetch(:connection_object).host
+ end
+ end
+
private
+
+ attr_reader :preferred_fallback_connection
+
+ # See #using_preferred_fallback_host? for context
+ def set_preferred_fallback_connection(connection)
+ @preferred_fallback_connection = if connection == @connection
+ # If the succeeded connection is in fact the primary connection (tried after a failed fallback)
+ # then clear the preferred fallback connection
+ nil
+ else
+ {
+ expires_at: Time.now + options.fetch(:fallback_retry_timeout),
+ connection_object: connection,
+ }
+ end
+ end
+
+ def get_preferred_fallback_connection_object
+ preferred_fallback_connection.fetch(:connection_object) if using_preferred_fallback_host?
+ end
+
def raw_request(method, path, params = {}, options = {})
options = options.clone
if options.delete(:disable_automatic_reauthorize) == true
send_request(method, path, params, options)
else
- reauthorize_on_authorisation_failure do
+ reauthorize_on_authorization_failure do
send_request(method, path, params, options)
end
end
end
@@ -441,14 +534,26 @@
requested_at = Time.now
retry_count = 0
retry_sequence_id = nil
request_id = SecureRandom.urlsafe_base64(10) if add_request_ids
+ preferred_fallback_connection_for_first_request = get_preferred_fallback_connection_object
+
begin
- use_fallback = can_fallback_to_alternate_ably_host? && retry_count > 0
+ use_fallback = can_fallback_to_alternate_ably_host? && (retry_count > 0)
- connection(use_fallback: use_fallback).send(method, path, params) do |request|
+ conn = if preferred_fallback_connection_for_first_request
+ case retry_count
+ when 0
+ preferred_fallback_connection_for_first_request
+ when 1
+ # Ensure the root host is used first if the preferred fallback fails, see #RSC15f
+ connection(use_fallback: false)
+ end
+ end || connection(use_fallback: use_fallback) # default to normal connection selection process if not preferred connection set
+
+ conn.send(method, path, params) do |request|
if add_request_ids
request.params[:request_id] = request_id
request.options.context = {} if request.options.context.nil?
request.options.context[:request_id] = request_id
end
@@ -460,44 +565,48 @@
end
end
end
end.tap do
if retry_count > 0
- logger.warn do
+ retry_log_severity = log_retries_as_info ? :info : :warn
+ logger.public_send(retry_log_severity) do
"Ably::Rest::Client - Request SUCCEEDED after #{retry_count} #{retry_count > 1 ? 'retries' : 'retry' } for" \
" #{method} #{path} #{params} (seq ##{retry_sequence_id}, time elapsed #{(Time.now.to_f - requested_at.to_f).round(2)}s)"
end
+ set_preferred_fallback_connection conn
end
end
rescue Faraday::TimeoutError, Faraday::ClientError, Ably::Exceptions::ServerError => error
retry_sequence_id ||= SecureRandom.urlsafe_base64(4)
time_passed = Time.now - requested_at
- if can_fallback_to_alternate_ably_host? && retry_count < max_retry_count && time_passed <= max_retry_duration
+ if can_fallback_to_alternate_ably_host? && (retry_count < max_retry_count) && (time_passed <= max_retry_duration)
retry_count += 1
- logger.warn { "Ably::Rest::Client - Retry #{retry_count} for #{method} #{path} #{params} as initial attempt failed (seq ##{retry_sequence_id}): #{error}" }
+ retry_log_severity = log_retries_as_info ? :info : :warn
+ logger.public_send(retry_log_severity) { "Ably::Rest::Client - Retry #{retry_count} for #{method} #{path} #{params} as initial attempt failed (seq ##{retry_sequence_id}): #{error}" }
retry
end
- logger.error do
+ retry_log_severity = log_retries_as_info ? :info : :error
+ logger.public_send(retry_log_severity) do
"Ably::Rest::Client - Request FAILED after #{retry_count} #{retry_count > 1 ? 'retries' : 'retry' } for" \
" #{method} #{path} #{params} (seq ##{retry_sequence_id}, time elapsed #{(Time.now.to_f - requested_at.to_f).round(2)}s)"
end
case error
when Faraday::TimeoutError
- raise Ably::Exceptions::ConnectionTimeout.new(error.message, nil, 80014, error, { request_id: request_id })
+ raise Ably::Exceptions::ConnectionTimeout.new(error.message, nil, Ably::Exceptions::Codes::CONNECTION_TIMED_OUT, error, { request_id: request_id })
when Faraday::ClientError
# request_id is also available in the request context
- raise Ably::Exceptions::ConnectionError.new(error.message, nil, 80000, error, { request_id: request_id })
+ raise Ably::Exceptions::ConnectionError.new(error.message, nil, Ably::Exceptions::Codes::CONNECTION_FAILED, error, { request_id: request_id })
else
raise error
end
end
end
- def reauthorize_on_authorisation_failure
+ def reauthorize_on_authorization_failure
yield
rescue Ably::Exceptions::TokenExpired => e
if auth.token_renewable?
auth.authorize
yield