diff --git a/UPDATING.md b/UPDATING.md index 85860259f..2d5c277bc 100644 --- a/UPDATING.md +++ b/UPDATING.md @@ -149,9 +149,68 @@ end `Ably::PubSub::Http::Channels#release` is unchanged. +### Protocol version 5 + +2.0 uses Ably protocol version 5 (previously 2). This changes the result of `publish` and the shape of `stats`, as +described below, and enables `update_message`, `delete_message` and `append_message` on HTTP and realtime channels. + +### Publish results + +`publish` now returns an `Ably::Models::PublishResult`. Its `serials` array holds one serial per published message, +in publish order; an entry is `nil` if the message was discarded by a conflation rule. + +```ruby +# 1.x +published = http_channel.publish('event', 'payload') # => true + +# 2.0 +result = http_channel.publish('event', 'payload') # => Ably::Models::PublishResult +result.serials # => [""] +``` + +On a realtime channel, the deferrable returned by `publish` now succeeds with a `PublishResult` instead of the +published `Message` (or `Array`): + +```ruby +# 1.x +realtime_channel.publish('event', 'payload') do |message| + message.name # => "event" +end + +# 2.0 +realtime_channel.publish('event', 'payload') do |result| + result.serials # => [""] +end +``` + +An HTTP publish that fails still raises an exception, so code that checked the `Boolean` result for success can drop +the check. + +### Stats + +`Ably::Models::Stats` follows the Ably spec v2.1+ (TS12) format. The nested accessors (`all`, `inbound`, `outbound`, +`persisted`, `connections`, `channels`, `api_requests`, `token_requests`) are replaced by `entries`, a flat `Hash` keyed +by dot-separated stat names: + +```ruby +stat = client.stats.items.first + +# 1.x +stat.all.all.count +stat.interval_granularity + +# 2.0 +stat.entries['messages.all.all.count'] +stat.unit +``` + +`Stats` also gains `in_progress`, `schema` and `app_id`. The `Stats::MessageTraffic`, `Stats::MessageTypes`, +`Stats::MessageCount`, `Stats::ConnectionTypes`, `Stats::RequestCount` and `Stats::ResourceCount` classes have been +removed. + ### Unchanged -Everything after construction: +Everything after construction, apart from the `publish` result and `Stats` changes above: ```ruby client = Ably::PubSub::Server.create_http_client(key: ENV['ABLY_API_KEY']) diff --git a/core/lib/ably/models/message.rb b/core/lib/ably/models/message.rb index f1d520d7e..82a89e6e7 100644 --- a/core/lib/ably/models/message.rb +++ b/core/lib/ably/models/message.rb @@ -26,6 +26,20 @@ class Message include Ably::Modules::Encodeable include Ably::Modules::ModelCommon include Ably::Modules::SafeDeferrable if defined?(Ably::PubSub::Realtime) + extend Ably::Modules::Enum + + # Describes the possible actions for a message. + # + # @spec TM5 + # + ACTION = ruby_enum('ACTION', + :message_create, # 0 + :message_update, # 1 + :message_delete, # 2 + :meta, # 3 + :message_summary, # 4 + :message_append, # 5 + ) # Statically register a default set of encoders for this class Ably::Models::MessageEncoders.register_default_encoders self @@ -129,10 +143,63 @@ def timestamp end end + # The action type of this message. + # + # @spec TM2j + # + # @return [ACTION, nil] + # + def action + ACTION(attributes[:action]) if attributes[:action] + end + + # An opaque string that uniquely identifies the message within a channel. + # + # @spec TM2r + # + # @return [String, nil] + # + def serial + attributes[:serial] + end + + # Version information about this message. + # + # @spec TM2s + # + # @return [Hash, nil] + # + def version + attributes[:version] + end + + # Timestamp of when the message was created. + # + # @return [Time, nil] + # + def created_at + as_time_from_epoch(attributes[:created_at]) if attributes[:created_at] + end + + # Timestamp of when the message was last updated. + # + # @return [Time, nil] + # + def updated_at + as_time_from_epoch(attributes[:updated_at]) if attributes[:updated_at] + end + def attributes @attributes end + # Return a JSON ready object from the underlying #attributes using Ably naming conventions for keys + def as_json(*args) + attributes.dup.tap do |message| + message['action'] = action.to_i if attributes[:action] + end.as_json.reject { |key, val| val.nil? } + end + def to_json(*args) as_json(*args).tap do |message| decode_binary_data_before_to_json message @@ -212,7 +279,7 @@ def raw_hash_object end def set_attributes_object(new_attributes) - @attributes = IdiomaticRubyWrapper(new_attributes.clone, stop_at: [:data, :extras]) + @attributes = IdiomaticRubyWrapper(new_attributes.clone, stop_at: [:data, :extras, :version]) end def logger diff --git a/core/lib/ably/models/message_operation.rb b/core/lib/ably/models/message_operation.rb new file mode 100644 index 000000000..3a1f2de2a --- /dev/null +++ b/core/lib/ably/models/message_operation.rb @@ -0,0 +1,53 @@ +module Ably::Models + # Represents the operation metadata for update, delete, or append message operations. + # + # @spec MOP + # + class MessageOperation + include Ably::Modules::ModelCommon + + # @param attributes [Hash] + # @option attributes [String] :client_id The client ID of the user performing the operation (MOP2a) + # @option attributes [String] :description An optional human-readable description of the operation (MOP2b) + # @option attributes [Hash] :metadata Arbitrary key-value metadata for the operation (MOP2c) + # + def initialize(attributes = {}) + @hash_object = IdiomaticRubyWrapper(attributes || {}, stop_at: [:metadata]) + @hash_object.freeze + end + + # The client ID of the user performing the operation. + # + # @spec MOP2a + # + # @return [String, nil] + # + def client_id + attributes[:client_id] + end + + # An optional human-readable description of the operation. + # + # @spec MOP2b + # + # @return [String, nil] + # + def description + attributes[:description] + end + + # Arbitrary key-value metadata for the operation. + # + # @spec MOP2c + # + # @return [Hash, nil] + # + def metadata + attributes[:metadata] + end + + def attributes + @hash_object + end + end +end diff --git a/core/lib/ably/models/protocol_message.rb b/core/lib/ably/models/protocol_message.rb index dc2a277c4..07a7e042f 100644 --- a/core/lib/ably/models/protocol_message.rb +++ b/core/lib/ably/models/protocol_message.rb @@ -171,6 +171,19 @@ def params @params ||= attributes[:params].to_h end + # The res field from ACK protocol messages (protocol v5+), containing publish results. + # Each entry corresponds to an ACK'd ProtocolMessage and contains a serials array + # with one entry per message in that ProtocolMessage. + # + # @spec TR4s + # + # @return [Array, nil] + # + # @api private + def res + attributes[:res] + end + def flags Integer(attributes[:flags]) rescue TypeError diff --git a/core/lib/ably/models/publish_result.rb b/core/lib/ably/models/publish_result.rb new file mode 100644 index 000000000..897f16efc --- /dev/null +++ b/core/lib/ably/models/publish_result.rb @@ -0,0 +1,32 @@ +module Ably::Models + # Contains the result of a publish operation. + # + # @spec RSL1n + # + class PublishResult + include Ably::Modules::ModelCommon + + # @param attributes [Hash] + # @option attributes [Array] :serials An array of nullable strings corresponding 1:1 + # to the published messages. Each serial identifies the message on the channel. + # + def initialize(attributes = {}) + @hash_object = IdiomaticRubyWrapper(attributes || {}, stop_at: [:serials]) + @hash_object.freeze + end + + # An array of serial strings (or nil entries), corresponding 1:1 to the published messages. + # + # @spec RSL1n + # + # @return [Array] + # + def serials + attributes[:serials] || [] + end + + def attributes + @hash_object + end + end +end diff --git a/core/lib/ably/models/stats.rb b/core/lib/ably/models/stats.rb index f7d919998..c13c8c27f 100644 --- a/core/lib/ably/models/stats.rb +++ b/core/lib/ably/models/stats.rb @@ -1,5 +1,3 @@ -require 'ably/models/stats_types' - module Ably::Models # Convert stat argument to a {Stats} object # @@ -17,16 +15,15 @@ def self.Stats(stat) # A class representing an individual statistic for a specified {#interval_id} # + # @spec TS12 + # class Stats include Ably::Modules::ModelCommon extend Ably::Modules::Enum # Describes the interval unit over which statistics are gathered. # - # MINUTE Interval unit over which statistics are gathered as minutes. - # HOUR Interval unit over which statistics are gathered as hours. - # DAY Interval unit over which statistics are gathered as days. - # MONTH Interval unit over which statistics are gathered as months. + # @spec TS12c # GRANULARITY = ruby_enum('GRANULARITY', :minute, @@ -103,123 +100,83 @@ def expected_length(format) # {Stats} initializer # - # @param hash_object [Hash] object with the underlying stat details + # @param hash_object [Hash] object with the underlying stat details # def initialize(hash_object) - @raw_hash_object = hash_object + @raw_hash_object = hash_object set_attributes_object hash_object end - # A {Ably::Models::Stats::MessageTypes} object containing the aggregate count of all message stats. + # The interval ID for this stats object. # - # @spec TS12e - # - # @return [Stats::MessageTypes] - # - def all - @all ||= Stats::MessageTypes.new(attributes[:all]) - end - - # A {Ably::Models::Stats::MessageTraffic} object containing the aggregate count of inbound message stats. - # - # @spec TS12f - # - # @return [Ably::Models::Stats::MessageTraffic] - # - def inbound - @inbound ||= Stats::MessageTraffic.new(attributes[:inbound]) - end - - # A {Ably::Models::Stats::MessageTraffic} object containing the aggregate count of outbound message stats. - # - # @spec TS12g - # - # @return [Ably::Models::Stats::MessageTraffic] - # - def outbound - @outbound ||= Stats::MessageTraffic.new(attributes[:outbound]) - end - - # A {Ably::Models::Stats::MessageTraffic} object containing the aggregate count of persisted message stats. - # - # @spec TS12h + # @spec TS12a # - # @return [Ably::Models::Stats::MessageTraffic] + # @return [String] # - def persisted - @persisted ||= Stats::MessageTypes.new(attributes[:persisted]) + def interval_id + attributes.fetch(:interval_id) end - # A {Ably::Models::Stats::ConnectionTypes} object containing a breakdown of connection related stats, such as min, mean and peak connections. + # Represents the intervalId as a time object. # - # @spec TS12i + # @spec TS12p # - # @return [Ably::Models::Stats::ConnectionTypes] + # @return [Time] # - def connections - @connections ||= Stats::ConnectionTypes.new(attributes[:connections]) + def interval_time + self.class.from_interval_id(interval_id) end - # A {Ably::Models::Stats::ResourceCount} object containing a breakdown of connection related stats, such as min, mean and peak connections. + # The unit of the interval, as provided by the API response. # - # @spec TS12j + # @spec TS12c # - # @return [Ably::Models::Stats::ResourceCount] + # @return [String] # - def channels - @channels ||= Stats::ResourceCount.new(attributes[:channels]) + def unit + attributes[:unit] end - # A {Ably::Models::Stats::RequestCount} object containing a breakdown of API Requests. + # For entries that are still in progress, the last sub-interval included. # - # @spec TS12k + # @spec TS12q # - # @return [Ably::Models::Stats::RequestCount] + # @return [String, nil] # - def api_requests - @api_requests ||= Stats::RequestCount.new(attributes[:api_requests]) + def in_progress + attributes[:in_progress] end - # A {Ably::Models::Stats::RequestCount} object containing a breakdown of Ably Token requests. + # A flat dictionary of statistics entries with dot-separated keys. # - # @spec TS12l + # @spec TS12r # - # @return [Ably::Models::Stats::RequestCount] + # @return [Hash] # - def token_requests - @token_requests ||= Stats::RequestCount.new(attributes[:token_requests]) + def entries + raw_entries = raw_hash_object[:entries] || raw_hash_object['entries'] + return {} unless raw_entries + raw_entries.is_a?(Hash) ? raw_entries : {} end - # The UTC time at which the time period covered begins. If unit is set to minute this will be in - # the format YYYY-mm-dd:HH:MM, if hour it will be YYYY-mm-dd:HH, if day it will be YYYY-mm-dd:00 - # and if month it will be YYYY-mm-01:00. + # The JSON schema URI for this stats object. # - # @spec TS12a + # @spec TS12s # # @return [String] # - def interval_id - attributes.fetch(:interval_id) + def schema + attributes[:schema] end - # Represents the intervalId as a time object. + # The Ably application ID this stats object relates to. # - # @spec TS12b + # @spec TS12t # - # @return [Time] - # - def interval_time - self.class.from_interval_id(interval_id) - end - - # The length of the interval the stats span. Values will be a [StatsIntervalGranularity]{@link StatsIntervalGranularity}. - # - # @spec TS12c - # - # @return [GRANULARITY] The granularity of the interval for the stat such as :day, :hour, :minute, see {GRANULARITY} + # @return [String] # - def interval_granularity - self.class.granularity_from_interval_id(interval_id) + def app_id + attributes[:app_id] end def attributes diff --git a/core/lib/ably/models/stats_types.rb b/core/lib/ably/models/stats_types.rb deleted file mode 100644 index 54eaa8672..000000000 --- a/core/lib/ably/models/stats_types.rb +++ /dev/null @@ -1,107 +0,0 @@ -module Ably::Models - class Stats - # StatsStruct is a basic Struct like class that allows methods to be defined - # on the class that will be retuned co-erced objects from the underlying hash used to - # initialize the object. - # - # This class provides a concise way to create classes that have fixed attributes and types - # - # @example - # class MessageCount < StatsStruct - # coerce_attributes :count, :data, into: Integer - # end - # - # @api private - # - class StatsStruct - class << self - def coerce_attributes(*attributes) - options = attributes.pop - raise ArgumentError, 'Expected attribute into: within options hash' unless options.kind_of?(Hash) && options[:into] - - @type_klass = options[:into] - setup_attribute_methods attributes - end - - def type_klass - @type_klass - end - - private - def setup_attribute_methods(attributes) - attributes.each do |attr| - define_method(attr) do - # Lazy load the co-erced value only when accessed - unless instance_variable_defined?("@#{attr}") - instance_variable_set "@#{attr}", self.class.type_klass.new(hash[attr.to_sym]) - end - instance_variable_get("@#{attr}") - end - end - end - end - - attr_reader :hash - - def initialize(hash) - @hash = hash || {} - end - end - - # IntegerDefaultZero will always return an Integer object and will default to value 0 unless truthy - # - # @api private - # - class IntegerDefaultZero - def self.new(value) - (value && value.to_i) || 0 - end - end - - # MessageCount contains aggregate counts for messages and data transferred - # - # @spec TS5a, TS5b - # - class MessageCount < StatsStruct - coerce_attributes :count, :data, into: IntegerDefaultZero - end - - # RequestCount contains aggregate counts for requests made - # - # @spec TS8a, TS8b, TS8c - # - class RequestCount < StatsStruct - coerce_attributes :succeeded, :failed, :refused, into: IntegerDefaultZero - end - - # ResourceCount contains aggregate data for usage of a resource in a specific scope - # - class ResourceCount < StatsStruct - coerce_attributes :opened, :peak, :mean, :min, :refused, into: IntegerDefaultZero - end - - # ConnectionTypes contains a breakdown of summary stats data for different (TLS vs non-TLS) connection types - # - # @spec TS4a, TS4b, TS4c - # - class ConnectionTypes < StatsStruct - coerce_attributes :tls, :plain, :all, into: ResourceCount - end - - # MessageTypes contains a breakdown of summary stats data for different (message vs presence) message types - # - # @spec TS6a, TS6b, TS6c - # - class MessageTypes < StatsStruct - coerce_attributes :messages, :presence, :all, into: MessageCount - end - - # MessageTraffic contains a breakdown of summary stats data for traffic over various transport types - # - # @spec TS7a, TS7b, TS7c, TS7d - # - class MessageTraffic < StatsStruct - coerce_attributes :realtime, :rest, :webhook, :all, into: MessageTypes - end - end -end diff --git a/core/lib/ably/models/update_delete_result.rb b/core/lib/ably/models/update_delete_result.rb new file mode 100644 index 000000000..3ca109490 --- /dev/null +++ b/core/lib/ably/models/update_delete_result.rb @@ -0,0 +1,32 @@ +module Ably::Models + # Contains the result of an update or delete message operation. + # + # @spec UDR + # + class UpdateDeleteResult + include Ably::Modules::ModelCommon + + # @param attributes [Hash] + # @option attributes [String, nil] :version_serial The new version serial string of the updated or deleted message. + # Will be nil if the message was superseded by a subsequent update before it could be published. + # + def initialize(attributes = {}) + @hash_object = IdiomaticRubyWrapper(attributes || {}) + @hash_object.freeze + end + + # The version serial of the updated or deleted message, or nil if superseded. + # + # @spec UDR2a + # + # @return [String, nil] + # + def version_serial + attributes[:version_serial] + end + + def attributes + @hash_object + end + end +end diff --git a/core/lib/ably/pubsub/http/channel.rb b/core/lib/ably/pubsub/http/channel.rb index f483ee608..6e910cebd 100644 --- a/core/lib/ably/pubsub/http/channel.rb +++ b/core/lib/ably/pubsub/http/channel.rb @@ -47,7 +47,7 @@ def initialize(client, name, channel_options = {}) # @param name [String, Array, Ably::Models::Message, nil] The event name of the message to publish, or an Array of [Ably::Model::Message] objects or [Hash] objects with +:name+ and +:data+ pairs, or a single Ably::Model::Message object # @param data [String, Array, Hash, nil] The message payload unless an Array of [Ably::Model::Message] objects passed in the first argument, in which case an optional hash of query parameters # @param attributes [Hash, nil] Optional additional message attributes such as :extras, :id, :client_id or :connection_id, applied when name attribute is nil or a string (Deprecated, will be removed in 2.0 in favour of constructing a Message object) - # @return [Boolean] true if the message was published, otherwise false + # @return [Ably::Models::PublishResult] A {Ably::Models::PublishResult} containing the serials of the published messages. # # @example # # Publish a single message with (name, data) form @@ -113,7 +113,54 @@ def publish(name, data = nil, attributes = {}) options = qs_params ? { qs_params: qs_params } : {} response = client.post("#{base_path}/publish", payload.length == 1 ? payload.first : payload, options) - [201, 204].include?(response.status) + parse_publish_result(response) + end + + # Updates a previously published message on the channel. + # + # @spec RSL15 + # + # @param message [Ably::Models::Message, Hash] A Message object or Hash containing a populated :serial field + # and the fields to update. + # @param operation [Hash, Ably::Models::MessageOperation, nil] Optional operation metadata containing + # :description and/or :metadata fields. + # @param params [Hash, nil] Optional parameters sent as part of the query string. + # + # @return [Ably::Models::UpdateDeleteResult] The result containing the version_serial. + # + def update_message(message, operation = nil, params = {}) + send_message_action(message, Ably::Models::Message::ACTION.MessageUpdate, operation, params) + end + + # Deletes a previously published message on the channel. + # + # @spec RSL15 + # + # @param message [Ably::Models::Message, Hash] A Message object or Hash containing a populated :serial field. + # @param operation [Hash, Ably::Models::MessageOperation, nil] Optional operation metadata containing + # :description and/or :metadata fields. + # @param params [Hash, nil] Optional parameters sent as part of the query string. + # + # @return [Ably::Models::UpdateDeleteResult] The result containing the version_serial. + # + def delete_message(message, operation = nil, params = {}) + send_message_action(message, Ably::Models::Message::ACTION.MessageDelete, operation, params) + end + + # Appends data to a previously published message on the channel. + # + # @spec RSL15 + # + # @param message [Ably::Models::Message, Hash] A Message object or Hash containing a populated :serial field + # and the data to append. + # @param operation [Hash, Ably::Models::MessageOperation, nil] Optional operation metadata containing + # :description and/or :metadata fields. + # @param params [Hash, nil] Optional parameters sent as part of the query string. + # + # @return [Ably::Models::UpdateDeleteResult] The result containing the version_serial. + # + def append_message(message, operation = nil, params = {}) + send_message_action(message, Ably::Models::Message::ACTION.MessageAppend, operation, params) end # Retrieves a {Ably::Models::PaginatedResult} object, containing an array of historical {Ably::Models::Message} @@ -179,6 +226,47 @@ def status private + def send_message_action(message, action_enum, operation = nil, params = {}) + message = Ably::Models::Message(message) + + raise Ably::Exceptions::InvalidRequest.new( + 'Message serial is required. Ensure the message has a serial field.' + ) unless message.serial + + update_attrs = message.as_json + update_attrs['action'] = action_enum.to_i + + if operation + op_hash = operation.respond_to?(:as_json) ? operation.as_json : operation + update_attrs['version'] = op_hash + end + + updated_message = Ably::Models::Message.new(update_attrs) + updated_message.encode client.encoders, options + + payload = updated_message.as_json + serial = message.serial + + request_options = params && !params.empty? ? { qs_params: params } : {} + response = client.patch( + "#{base_path}/messages/#{URI.encode_www_form_component(serial)}", + payload, + request_options + ) + + Ably::Models::UpdateDeleteResult.new(response.body || {}) + end + + def parse_publish_result(response) + body = response.body + if body.is_a?(Hash) + serials = body['serials'] || body[:serials] || [] + Ably::Models::PublishResult.new(serials: serials) + else + Ably::Models::PublishResult.new(serials: []) + end + end + def base_path "/channels/#{URI.encode_www_form_component(name)}" end diff --git a/core/lib/ably/pubsub/http/client.rb b/core/lib/ably/pubsub/http/client.rb index bbc8ee0b8..65f9c5e9b 100644 --- a/core/lib/ably/pubsub/http/client.rb +++ b/core/lib/ably/pubsub/http/client.rb @@ -359,6 +359,15 @@ def put(path, params, options = {}) raw_request(:put, path, params, options) end + # Perform an HTTP PATCH request to the API using configured authentication + # + # @return [Faraday::Response] + # + # @api private + def patch(path, params, options = {}) + raw_request(:patch, path, params, options) + end + # Perform an HTTP DELETE request to the API using configured authentication # # @return [Faraday::Response] diff --git a/core/lib/ably/pubsub/realtime/channel.rb b/core/lib/ably/pubsub/realtime/channel.rb index a5de314f0..85ac0a2e8 100644 --- a/core/lib/ably/pubsub/realtime/channel.rb +++ b/core/lib/ably/pubsub/realtime/channel.rb @@ -214,6 +214,56 @@ def publish(name, data = nil, attributes = {}, &success_block) end end + # Updates a previously published message on the channel. A callback may optionally be passed in to this + # call to be notified of success or failure of the operation. + # + # @spec RTL32 + # + # @param message [Ably::Models::Message, Hash] A Message object or Hash containing a populated :serial field + # and the fields to update. + # @param operation [Hash, Ably::Models::MessageOperation, nil] Optional operation metadata. + # @param params [Hash, nil] Optional parameters sent as part of the protocol message. + # + # @yield [Ably::Models::UpdateDeleteResult] On success, calls the block with the result containing version_serial. + # @return [Ably::Util::SafeDeferrable] Deferrable that supports both success (callback) and failure (errback) callbacks + # + def update_message(message, operation = nil, params = {}, &success_block) + send_message_action(message, Ably::Models::Message::ACTION.MessageUpdate, operation, params, &success_block) + end + + # Deletes a previously published message on the channel. A callback may optionally be passed in to this + # call to be notified of success or failure of the operation. + # + # @spec RTL32 + # + # @param message [Ably::Models::Message, Hash] A Message object or Hash containing a populated :serial field. + # @param operation [Hash, Ably::Models::MessageOperation, nil] Optional operation metadata. + # @param params [Hash, nil] Optional parameters sent as part of the protocol message. + # + # @yield [Ably::Models::UpdateDeleteResult] On success, calls the block with the result containing version_serial. + # @return [Ably::Util::SafeDeferrable] Deferrable that supports both success (callback) and failure (errback) callbacks + # + def delete_message(message, operation = nil, params = {}, &success_block) + send_message_action(message, Ably::Models::Message::ACTION.MessageDelete, operation, params, &success_block) + end + + # Appends data to a previously published message on the channel. A callback may optionally be passed in to this + # call to be notified of success or failure of the operation. + # + # @spec RTL32 + # + # @param message [Ably::Models::Message, Hash] A Message object or Hash containing a populated :serial field + # and the data to append. + # @param operation [Hash, Ably::Models::MessageOperation, nil] Optional operation metadata. + # @param params [Hash, nil] Optional parameters sent as part of the protocol message. + # + # @yield [Ably::Models::UpdateDeleteResult] On success, calls the block with the result containing version_serial. + # @return [Ably::Util::SafeDeferrable] Deferrable that supports both success (callback) and failure (errback) callbacks + # + def append_message(message, operation = nil, params = {}, &success_block) + send_message_action(message, Ably::Models::Message::ACTION.MessageAppend, operation, params, &success_block) + end + # Registers a listener for messages on this channel. The caller supplies a listener function, which is called # each time one or more messages arrives on the channel. A callback may optionally be passed in to this call # to be notified of success or failure of the channel {Ably::PubSub::Realtime::Channel#attach} operation. @@ -406,6 +456,59 @@ def need_reattach? private + def send_message_action(message, action_enum, operation = nil, params = {}, &success_block) + if suspended? || failed? + error = Ably::Exceptions::ChannelInactive.new("Cannot send message action on a channel in state #{state}") + return Ably::Util::SafeDeferrable.new_and_fail_immediately(logger, error) + end + + if !connection.can_publish_messages? + error = Ably::Exceptions::MessageQueueingDisabled.new( + "Message action cannot be sent. Client is not allowed to queue messages when connection is in state #{connection.state}" + ) + return Ably::Util::SafeDeferrable.new_and_fail_immediately(logger, error) + end + + message = Ably::Models::Message(message) + + unless message.serial + error = Ably::Exceptions::InvalidRequest.new('Message serial is required. Ensure the message has a serial field.') + return Ably::Util::SafeDeferrable.new_and_fail_immediately(logger, error) + end + + update_attrs = message.as_json + update_attrs['action'] = action_enum.to_i + + if operation + op_hash = operation.respond_to?(:as_json) ? operation.as_json : operation + update_attrs['version'] = op_hash + end + + updated_message = Ably::Models::Message.new(update_attrs) + updated_message.encode(client.encoders, options) do |encode_error, error_message| + client.logger.error error_message + end + + pm_params = { action: Ably::Models::ProtocolMessage::ACTION.Message.to_i, channel: name, messages: [updated_message] } + pm_params[:params] = params.transform_values(&:to_s) if params && !params.empty? + + connection.send_protocol_message(pm_params) + + Ably::Util::SafeDeferrable.new(logger).tap do |deferrable| + updated_message.callback do |result| + if result.is_a?(Ably::Models::UpdateDeleteResult) + deferrable.succeed result + else + deferrable.succeed Ably::Models::UpdateDeleteResult.new(version_serial: nil) + end + end + updated_message.errback do |error| + deferrable.fail error + end + deferrable.callback(&success_block) if block_given? + end + end + def setup_event_handlers __incoming_msgbus__.subscribe(:message) do |message| message.decode(client.encoders, options) do |encode_error, error_message| diff --git a/core/lib/ably/pubsub/realtime/channel/publisher.rb b/core/lib/ably/pubsub/realtime/channel/publisher.rb index 12dbfe71a..e6e01680c 100644 --- a/core/lib/ably/pubsub/realtime/channel/publisher.rb +++ b/core/lib/ably/pubsub/realtime/channel/publisher.rb @@ -52,13 +52,18 @@ def deferrable_for_multiple_messages(messages) expected_deliveries = messages.count actual_deliveries = 0 failed = false + results = Array.new(messages.count) Ably::Util::SafeDeferrable.new(logger).tap do |deferrable| - messages.each do |message| - message.callback do + messages.each_with_index do |message, index| + message.callback do |result| next if failed + results[index] = result actual_deliveries += 1 - deferrable.succeed messages if actual_deliveries == expected_deliveries + if actual_deliveries == expected_deliveries + all_serials = results.flat_map(&:serials) + deferrable.succeed Ably::Models::PublishResult.new(serials: all_serials) + end end message.errback do |error| next if failed diff --git a/core/lib/ably/pubsub/realtime/client/incoming_message_dispatcher.rb b/core/lib/ably/pubsub/realtime/client/incoming_message_dispatcher.rb index 8adbcea60..a0197c085 100644 --- a/core/lib/ably/pubsub/realtime/client/incoming_message_dispatcher.rb +++ b/core/lib/ably/pubsub/realtime/client/incoming_message_dispatcher.rb @@ -178,9 +178,12 @@ def process_connected_update_message(protocol_message) end def ack_pending_queue_for_message_serial(ack_protocol_message) + res_index = 0 drop_pending_queue_from_ack(ack_protocol_message) do |protocol_message| - ack_messages protocol_message.messages + publish_result = ack_protocol_message.res[res_index] if ack_protocol_message.res + ack_messages protocol_message.messages, publish_result ack_messages protocol_message.presence + res_index += 1 end end @@ -191,11 +194,43 @@ def nack_pending_queue_for_message_serial(nack_protocol_message) end end - def ack_messages(messages) - messages.each do |message| + def ack_messages(messages, publish_result = nil) + messages.each_with_index do |message, index| logger.debug { "Calling ACK success callbacks for #{message.class.name} - #{message.to_json}" } - message.succeed message + if publish_result && message.respond_to?(:action) && message.action && + message.action.match_any?( + Ably::Models::Message::ACTION.MessageUpdate, + Ably::Models::Message::ACTION.MessageDelete, + Ably::Models::Message::ACTION.MessageAppend + ) + serials = extract_serials(publish_result) + version_serial = serials[index] if serials + result = Ably::Models::UpdateDeleteResult.new(version_serial: version_serial) + message.succeed result + elsif publish_result && message.is_a?(Ably::Models::Message) + serials = extract_serials(publish_result) + serial = serials ? serials[index] : nil + result = Ably::Models::PublishResult.new(serials: [serial]) + message.succeed result + else + message.succeed message + end + end + end + + def extract_serials(publish_result) + return nil unless publish_result + + if publish_result.is_a?(Hash) + publish_result['serials'] || publish_result[:serials] + elsif publish_result.respond_to?(:[]) + publish_result[:serials] + else + nil end + rescue StandardError => e + logger.warn { "Failed to extract serials from publish_result (#{publish_result.class}): #{e.message}" } + nil end def nack_messages(messages, protocol_message) diff --git a/core/lib/ably/version.rb b/core/lib/ably/version.rb index ed629fa87..897b7236e 100644 --- a/core/lib/ably/version.rb +++ b/core/lib/ably/version.rb @@ -3,5 +3,5 @@ module Ably # The level of compatibility with the Ably service that this SDK supports. # Also referred to as the 'wire protocol version'. # spec : CSV2 - PROTOCOL_VERSION = '2' + PROTOCOL_VERSION = '5' end diff --git a/spec/acceptance/http/channel_spec.rb b/spec/acceptance/http/channel_spec.rb index 31bbb9279..ee4026f34 100644 --- a/spec/acceptance/http/channel_spec.rb +++ b/spec/acceptance/http/channel_spec.rb @@ -20,8 +20,8 @@ let(:data) { 'woop!' } context 'with name and data arguments' do - it 'publishes the message and return true indicating success' do - expect(channel.publish(name, data)).to eql(true) + it 'publishes the message and returns a PublishResult' do + expect(channel.publish(name, data)).to be_a(Ably::Models::PublishResult) expect(channel.history.items.first.name).to eql(name) expect(channel.history.items.first.data).to eql(data) end @@ -29,8 +29,8 @@ context 'and additional attributes' do let(:client_id) { random_str } - it 'publishes the message with the attributes and return true indicating success' do - expect(channel.publish(name, data, client_id: client_id)).to eql(true) + it 'publishes the message with the attributes and returns a PublishResult' do + expect(channel.publish(name, data, client_id: client_id)).to be_a(Ably::Models::PublishResult) expect(channel.history.items.first.client_id).to eql(client_id) end end @@ -43,9 +43,9 @@ it 'publishes the message without a client_id' do expect(client).to receive(:post). with("/channels/#{channel_name}/publish", hash_excluding(client_id: client_id), {}). - and_return(double('response', status: 201)) + and_return(double('response', status: 201, body: {})) - expect(channel.publish(name, data)).to eql(true) + expect(channel.publish(name, data)).to be_a(Ably::Models::PublishResult) end it 'expects a client_id to be added by the realtime service' do @@ -66,7 +66,7 @@ expect(messages.sum(&:size) < Ably::PubSub::Http::Client::MAX_MESSAGE_SIZE).to eq(true) expect(client).to receive(:post).once.and_call_original - expect(channel.publish(messages)).to eql(true) + expect(channel.publish(messages)).to be_a(Ably::Models::PublishResult) expect(channel.history.items.map(&:name)).to match_array(messages.map { |message| message[:name] }) expect(channel.history.items.map(&:data)).to match_array(messages.map { |message| message[:data] }) end @@ -89,7 +89,7 @@ it 'publishes an array of messages in one HTTP request' do expect(messages.sum &:size).to eq(130) expect(client).to receive(:post).once.and_call_original - expect(channel.publish(messages)).to eql(true) + expect(channel.publish(messages)).to be_a(Ably::Models::PublishResult) expect(channel.history.items.map(&:name)).to match_array(messages.map(&:name)) expect(channel.history.items.map(&:data)).to match_array(messages.map(&:data)) end @@ -127,7 +127,7 @@ it 'publishes an array of messages in one HTTP request' do expect(messages.sum &:size).to eq(130) expect(client).to receive(:post).once.and_call_original - expect(channel.publish(messages)).to eql(true) + expect(channel.publish(messages)).to be_a(Ably::Models::PublishResult) expect(channel.history.items.map(&:name)).to match_array(messages.map(&:name)) expect(channel.history.items.map(&:data)).to match_array(messages.map(&:data)) end @@ -157,7 +157,7 @@ it 'publishes the message' do expect(client).to receive(:post).once.and_call_original - expect(channel.publish(message)).to eql(true) + expect(channel.publish(message)).to be_a(Ably::Models::PublishResult) expect(channel.history.items.first.name).to eql(name) end end @@ -201,7 +201,7 @@ it 'publishes the message without a name attribute in the payload' do expect(client).to receive(:post).with(anything, { "data" => data }, {}).once.and_call_original - expect(channel.publish(nil, data)).to eql(true) + expect(channel.publish(nil, data)).to be_a(Ably::Models::PublishResult) expect(channel.history.items.first.name).to be_nil expect(channel.history.items.first.data).to eql(data) end @@ -212,7 +212,7 @@ it 'publishes the message without a data attribute in the payload' do expect(client).to receive(:post).with(anything, { "name" => name }, {}).once.and_call_original - expect(channel.publish(name)).to eql(true) + expect(channel.publish(name)).to be_a(Ably::Models::PublishResult) expect(channel.history.items.first.name).to eql(name) expect(channel.history.items.first.data).to be_nil end @@ -223,7 +223,7 @@ it 'publishes the message without any attributes in the payload' do expect(client).to receive(:post).with(anything, {}, {}).once.and_call_original - expect(channel.publish(nil)).to eql(true) + expect(channel.publish(nil)).to be_a(Ably::Models::PublishResult) expect(channel.history.items.first.name).to be_nil expect(channel.history.items.first.data).to be_nil end diff --git a/spec/acceptance/http/client_spec.rb b/spec/acceptance/http/client_spec.rb index 2aae4b959..c9d0d3ba0 100644 --- a/spec/acceptance/http/client_spec.rb +++ b/spec/acceptance/http/client_spec.rb @@ -1067,14 +1067,14 @@ def encode64(text) it 'sends a protocol version and lib version header (#G4, #RSC7a, #RSC7b)' do response = client.channels.get('foo').publish("event") - expect(response).to eql true + expect(response).to be_a(Ably::Models::PublishResult) expect(publish_message_stub).to have_been_requested if agent.nil? expect(publish_message_stub.to_s).to include("'Ably-Agent'=>'#{Ably::AGENT}'") - expect(publish_message_stub.to_s).to include("'X-Ably-Version'=>'2'") + expect(publish_message_stub.to_s).to include("'X-Ably-Version'=>'#{Ably::PROTOCOL_VERSION}'") else expect(publish_message_stub.to_s).to include("'Ably-Agent'=>'ably-ruby/1.1.1 ruby/3.1.1'") - expect(publish_message_stub.to_s).to include("'X-Ably-Version'=>'2'") + expect(publish_message_stub.to_s).to include("'X-Ably-Version'=>'#{Ably::PROTOCOL_VERSION}'") end end end diff --git a/spec/acceptance/http/encoders_spec.rb b/spec/acceptance/http/encoders_spec.rb index 8b86b96ff..8e8cebdb9 100644 --- a/spec/acceptance/http/encoders_spec.rb +++ b/spec/acceptance/http/encoders_spec.rb @@ -7,7 +7,7 @@ let(:client) { Ably::Internal.create_http_client(default_client_options.merge(protocol: protocol)) } let(:channel_options) { {} } let(:channel) { client.channel('test', channel_options) } - let(:response) { instance_double('Faraday::Response', status: 201) } + let(:response) { instance_double('Faraday::Response', status: 201, body: {}) } let(:cipher_params) { { key: Ably::Util::Crypto.generate_random_key(128), algorithm: 'aes', mode: 'cbc', key_length: 128 } } let(:crypto) { Ably::Util::Crypto.new(cipher_params) } diff --git a/spec/acceptance/http/message_spec.rb b/spec/acceptance/http/message_spec.rb index b1192c985..45fe4b248 100644 --- a/spec/acceptance/http/message_spec.rb +++ b/spec/acceptance/http/message_spec.rb @@ -381,7 +381,7 @@ def mock_for_two_publish_failures expect(message['encoding']).to eql(encrypted_encoding.gsub(%r{/base64$}, '')) expect(message['data']).to eql(encrypted_data_decoded) end - end.and_return(double('Response', status: 201)) + end.and_return(double('Response', status: 201, body: {})) encrypted_channel.publish 'example', encoded_data_decoded end diff --git a/spec/acceptance/http/stats_spec.rb b/spec/acceptance/http/stats_spec.rb index dc89a8db1..5fe393ff1 100644 --- a/spec/acceptance/http/stats_spec.rb +++ b/spec/acceptance/http/stats_spec.rb @@ -63,8 +63,8 @@ let(:subject) { client.stats(end: LAST_INTERVAL) } # end is needed to ensure no other tests have effected the stats let(:stat) { subject.items.first } - it 'uses the minute interval by default' do - expect(stat.interval_granularity).to eq(:minute) + it 'returns the unit from the JSON response' do + expect(stat.unit).to eq('minute') end end @@ -76,58 +76,62 @@ expect(subject.items.count).to eql(1) end - it 'returns zero value for any missing metrics' do - expect(stat.channels.refused).to eql(0) - expect(stat.outbound.webhook.all.count).to eql(0) + it 'returns entries as a flat hash (#TS12r)' do + expect(stat.entries).to be_a(Hash) + end + + it 'returns zero or nil for any missing entries' do + expect(stat.entries['channels.refused']).to be_nil + expect(stat.entries['messages.outbound.webhook.all.count']).to be_nil end it 'returns all aggregated message data' do - expect(stat.all.messages.count).to eql(70 + 40) # inbound + outbound - expect(stat.all.messages.data).to eql(7000 + 4000) # inbound + outbound + expect(stat.entries['messages.all.messages.count']).to eql(70 + 40) # inbound + outbound + expect(stat.entries['messages.all.messages.data']).to eql(7000 + 4000) # inbound + outbound end it 'returns inbound realtime all data' do - expect(stat.inbound.realtime.all.count).to eql(70) - expect(stat.inbound.realtime.all.data).to eql(7000) + expect(stat.entries['messages.inbound.realtime.all.count']).to eql(70) + expect(stat.entries['messages.inbound.realtime.all.data']).to eql(7000) end it 'returns inbound realtime message data' do - expect(stat.inbound.realtime.messages.count).to eql(70) - expect(stat.inbound.realtime.messages.data).to eql(7000) + expect(stat.entries['messages.inbound.realtime.messages.count']).to eql(70) + expect(stat.entries['messages.inbound.realtime.messages.data']).to eql(7000) end it 'returns outbound realtime all data' do - expect(stat.outbound.realtime.all.count).to eql(40) - expect(stat.outbound.realtime.all.data).to eql(4000) + expect(stat.entries['messages.outbound.realtime.all.count']).to eql(40) + expect(stat.entries['messages.outbound.realtime.all.data']).to eql(4000) end it 'returns persisted presence all data' do - expect(stat.persisted.all.count).to eql(20) - expect(stat.persisted.all.data).to eql(2000) + expect(stat.entries['messages.persisted.all.count']).to eql(20) + expect(stat.entries['messages.persisted.all.data']).to eql(2000) end it 'returns connections all data' do - expect(stat.connections.tls.peak).to eql(20) - expect(stat.connections.tls.opened).to eql(10) + expect(stat.entries['connections.all.peak']).to eql(20) + expect(stat.entries['connections.all.opened']).to eql(10) end - it 'returns channels all data' do - expect(stat.channels.peak).to eql(50) - expect(stat.channels.opened).to eql(30) + it 'returns channels data' do + expect(stat.entries['channels.peak']).to eql(50) + expect(stat.entries['channels.opened']).to eql(30) end it 'returns api_requests data' do - expect(stat.api_requests.succeeded).to eql(50) - expect(stat.api_requests.failed).to eql(10) + expect(stat.entries['apiRequests.all.succeeded']).to eql(110) # 50 apiRequests + 60 tokenRequests + expect(stat.entries['apiRequests.all.failed']).to eql(30) # 10 apiRequests + 20 tokenRequests end it 'returns token_requests data' do - expect(stat.token_requests.succeeded).to eql(60) - expect(stat.token_requests.failed).to eql(20) + expect(stat.entries['apiRequests.tokenRequests.succeeded']).to eql(60) + expect(stat.entries['apiRequests.tokenRequests.failed']).to eql(20) end - it 'returns stat objects with #interval_granularity equal to :minute' do - expect(stat.interval_granularity).to eq(:minute) + it 'returns stat objects with #unit equal to minute' do + expect(stat.unit).to eq('minute') end it 'returns stat objects with #interval_id matching :start' do @@ -145,14 +149,14 @@ let(:stat) { subject.items.first } it 'returns the first interval stats as stats are provided forwards from :start' do - expect(stat.inbound.realtime.all.count).to eql(first_inbound_realtime_count) + expect(stat.entries['messages.inbound.realtime.all.count']).to eql(first_inbound_realtime_count) end it 'returns 3 pages of stats' do expect(subject).to_not be_last page3 = subject.next.next expect(page3).to be_last - expect(page3.items.first.inbound.realtime.all.count).to eql(last_inbound_realtime_count) + expect(page3.items.first.entries['messages.inbound.realtime.all.count']).to eql(last_inbound_realtime_count) end end @@ -161,13 +165,13 @@ let(:stat) { subject.items.first } it 'returns the 3rd interval stats first as stats are provided backwards from :end' do - expect(stat.inbound.realtime.all.count).to eql(last_inbound_realtime_count) + expect(stat.entries['messages.inbound.realtime.all.count']).to eql(last_inbound_realtime_count) end it 'returns 3 pages of stats' do expect(subject).to_not be_last page3 = subject.next.next - expect(page3.items.first.inbound.realtime.all.count).to eql(first_inbound_realtime_count) + expect(page3.items.first.entries['messages.inbound.realtime.all.count']).to eql(first_inbound_realtime_count) end end @@ -177,8 +181,8 @@ context 'the REST API' do it 'defaults to direction :backwards' do - expect(stats.first.inbound.realtime.messages.count).to eql(70) # current minute - expect(stats.last.inbound.realtime.messages.count).to eql(50) # 2 minutes back + expect(stats.first.entries['messages.inbound.realtime.messages.count']).to eql(70) # current minute + expect(stats.last.entries['messages.inbound.realtime.messages.count']).to eql(50) # 2 minutes back end end end @@ -215,8 +219,8 @@ it 'should aggregate the stats for that period' do expect(subject.items.count).to eql(1) - expect(stat.all.messages.count).to eql(aggregate_messages_count) - expect(stat.all.messages.data).to eql(aggregate_messages_data) + expect(stat.entries['messages.all.messages.count']).to eql(aggregate_messages_count) + expect(stat.entries['messages.all.messages.data']).to eql(aggregate_messages_data) end end end diff --git a/spec/acceptance/realtime/auth_spec.rb b/spec/acceptance/realtime/auth_spec.rb index 7890fddc8..93a0718d3 100644 --- a/spec/acceptance/realtime/auth_spec.rb +++ b/spec/acceptance/realtime/auth_spec.rb @@ -1243,8 +1243,8 @@ def disconnect_transport(connection) forbidden_channel.publish('not-allowed').errback do |error| expect(error.code).to eql(40160) - allowed_channel.publish(message_name) do |message| - expect(message.name).to eql(message_name) + allowed_channel.publish(message_name) do |result| + expect(result).to be_a(Ably::Models::PublishResult) stop_reactor end end diff --git a/spec/acceptance/realtime/connection_spec.rb b/spec/acceptance/realtime/connection_spec.rb index 245966de4..c53ab90db 100644 --- a/spec/acceptance/realtime/connection_spec.rb +++ b/spec/acceptance/realtime/connection_spec.rb @@ -1996,7 +1996,7 @@ def self.available_states it 'sends the protocol version param v (#G4, #RTN2f)' do expect(EventMachine).to receive(:connect) do |host, port, transport, object, url| uri = URI.parse(url) - expect(CGI::parse(uri.query)['v'][0]).to eql('2') + expect(CGI::parse(uri.query)['v'][0]).to eql('5') stop_reactor end client diff --git a/spec/acceptance/realtime/message_spec.rb b/spec/acceptance/realtime/message_spec.rb index ee0da5597..c236307d8 100644 --- a/spec/acceptance/realtime/message_spec.rb +++ b/spec/acceptance/realtime/message_spec.rb @@ -25,10 +25,11 @@ it 'sends a String data payload' do channel.attach channel.on(:attached) do - channel.publish('test_event', payload) do |message| + channel.subscribe do |message| expect(message.data).to eql(payload) stop_reactor end + channel.publish('test_event', payload) end end diff --git a/spec/support/event_machine_helper.rb b/spec/support/event_machine_helper.rb index 0fa05e468..c1cd6f640 100644 --- a/spec/support/event_machine_helper.rb +++ b/spec/support/event_machine_helper.rb @@ -2,6 +2,26 @@ require 'rspec' require 'timeout' +# Prevent EM signal_loopbreak race condition from corrupting the entire +# process. When EM.defer is used and the reactor stops while a threadpool +# worker is mid-flight, the worker calls signal_loopbreak on a stopped +# reactor, raising RuntimeError. This poisons EM for the rest of the +# process, cascading failures to every subsequent test that touches EM. +# Swallowing the error is safe — signal_loopbreak is just a notification +# that there's work to process, and if EM isn't running there's nothing +# to notify. +module EventMachine + class << self + alias_method :original_signal_loopbreak, :signal_loopbreak + + def signal_loopbreak + original_signal_loopbreak + rescue RuntimeError + # EM not initialized — reactor was stopped between the check and the call + end + end +end + module RSpec module EventMachine extend self @@ -139,6 +159,21 @@ def wait_until(condition_block, &block) # Ensure EventMachine shutdown hooks are deregistered for every test EventMachine.instance_variable_set '@tails', [] end + + # Catch-all cleanup for ANY test that used EventMachine, whether via + # the :event_machine tag or by calling run_reactor directly. Without this, + # a crashed/timed-out reactor and stale client references leak into + # subsequent tests causing cascade failures. + config.after(:example) do + RSpec::EventMachine.realtime_clients.clear + begin + EventMachine.stop if EventMachine.reactor_running? + rescue RuntimeError + # EM can be in a corrupted state (e.g. signal_loopbreak failure) + # where reactor_running? returns true but stop raises. Swallow + # the error to prevent cascading failures across subsequent tests. + end + end end module RSpec diff --git a/spec/unit/http/channel_spec.rb b/spec/unit/http/channel_spec.rb index 928f98219..945f56404 100644 --- a/spec/unit/http/channel_spec.rb +++ b/spec/unit/http/channel_spec.rb @@ -2,11 +2,12 @@ require 'spec_helper' describe Ably::PubSub::Http::Channel do + let(:post_response) { instance_double('Faraday::Response', status: 201, body: { 'serials' => ['serial-001'] }) } let(:client) do instance_double( 'Ably::PubSub::Http::Client', encoders: [], - post: instance_double('Faraday::Response', status: 201), + post: post_response, idempotent_rest_publishing: false, max_message_size: max_message_size ) end @@ -91,7 +92,7 @@ let(:encoding) { Encoding::UTF_8 } it 'is permitted' do - expect(subject.publish(encoded_value, 'data')).to eql(true) + expect(subject.publish(encoded_value, 'data')).to be_a(Ably::Models::PublishResult) end end @@ -100,7 +101,7 @@ let(:encoding) { Encoding::UTF_8 } it 'is permitted' do - expect(subject.publish(encoded_value, 'data')).to eql(true) + expect(subject.publish(encoded_value, 'data')).to be_a(Ably::Models::PublishResult) end end @@ -108,7 +109,7 @@ let(:encoding) { Encoding::SHIFT_JIS } it 'is permitted' do - expect(subject.publish(encoded_value, 'data')).to eql(true) + expect(subject.publish(encoded_value, 'data')).to be_a(Ably::Models::PublishResult) end end @@ -116,7 +117,7 @@ let(:encoding) { Encoding::ASCII_8BIT } it 'is permitted' do - expect(subject.publish(encoded_value, 'data')).to eql(true) + expect(subject.publish(encoded_value, 'data')).to be_a(Ably::Models::PublishResult) end end @@ -148,7 +149,7 @@ context 'and a message size is 10 bytes' do it 'should send a message' do - expect(subject.publish('x' * 10, 'data')).to eq(true) + expect(subject.publish('x' * 10, 'data')).to be_a(Ably::Models::PublishResult) end end end @@ -164,10 +165,295 @@ context 'and a message size is 2 bytes' do it 'should send a message' do - expect(subject.publish('x' * 2, 'data')).to eq(true) + expect(subject.publish('x' * 2, 'data')).to be_a(Ably::Models::PublishResult) end end end end end + + describe '#publish returns PublishResult (#RSL1n)' do + context 'with serials in response body' do + let(:post_response) { instance_double('Faraday::Response', status: 201, body: { 'serials' => ['serial-abc', 'serial-def'] }) } + + it 'returns a PublishResult with serials' do + result = subject.publish('event', 'data') + expect(result).to be_a(Ably::Models::PublishResult) + expect(result.serials).to eql(['serial-abc', 'serial-def']) + end + end + + context 'with empty response body (204)' do + let(:post_response) { instance_double('Faraday::Response', status: 204, body: nil) } + + it 'returns a PublishResult with empty serials' do + result = subject.publish('event', 'data') + expect(result).to be_a(Ably::Models::PublishResult) + expect(result.serials).to eql([]) + end + end + + context 'with non-hash response body' do + let(:post_response) { instance_double('Faraday::Response', status: 201, body: '') } + + it 'returns a PublishResult with empty serials' do + result = subject.publish('event', 'data') + expect(result).to be_a(Ably::Models::PublishResult) + expect(result.serials).to eql([]) + end + end + end + + describe '#update_message (#RSL15)' do + let(:serial) { 'msg-serial-001' } + let(:patch_response) { instance_double('Faraday::Response', status: 200, body: { 'versionSerial' => 'v1-serial' }) } + let(:client) do + instance_double( + 'Ably::PubSub::Http::Client', + encoders: [], + post: instance_double('Faraday::Response', status: 201, body: { 'serials' => ['serial-001'] }), + patch: patch_response, + idempotent_rest_publishing: false, + max_message_size: max_message_size + ) + end + + context 'with a valid message containing serial' do + let(:message) { Ably::Models::Message.new(name: 'test', data: 'hello', serial: serial) } + + it 'sends a PATCH request' do + expect(client).to receive(:patch).with( + "/channels/#{channel_name}/messages/#{serial}", + hash_including('action' => 1), + {} + ).and_return(patch_response) + + subject.update_message(message) + end + + it 'returns an UpdateDeleteResult' do + result = subject.update_message(message) + expect(result).to be_a(Ably::Models::UpdateDeleteResult) + expect(result.version_serial).to eql('v1-serial') + end + + it 'sets action to MESSAGE_UPDATE' do + expect(client).to receive(:patch) do |_path, payload, _opts| + expect(payload['action']).to eq(Ably::Models::Message::ACTION.MessageUpdate.to_i) + patch_response + end + + subject.update_message(message) + end + end + + context 'with an operation parameter' do + let(:message) { Ably::Models::Message.new(name: 'test', serial: serial) } + let(:operation) { { description: 'Fixed typo', metadata: { 'reason' => 'correction' } } } + + it 'includes the operation as version in the payload' do + expect(client).to receive(:patch) do |_path, payload, _opts| + expect(payload['version']).to eq({ 'description' => 'Fixed typo', 'metadata' => { 'reason' => 'correction' } }) + patch_response + end + + subject.update_message(message, operation) + end + end + + context 'with a MessageOperation object' do + let(:message) { Ably::Models::Message.new(name: 'test', serial: serial) } + let(:operation) { Ably::Models::MessageOperation.new(description: 'Fixed typo') } + + it 'serializes the operation via as_json' do + expect(client).to receive(:patch) do |_path, payload, _opts| + expect(payload['version']).to be_a(Hash) + expect(payload['version']['description']).to eq('Fixed typo') + patch_response + end + + subject.update_message(message, operation) + end + end + + context 'without serial (#RSL15a)' do + let(:message) { Ably::Models::Message.new(name: 'test', data: 'hello') } + + it 'raises an InvalidRequest exception' do + expect { subject.update_message(message) }.to raise_error(Ably::Exceptions::InvalidRequest, /serial is required/) + end + end + + context 'with a Hash message' do + it 'converts to Message and validates serial' do + expect { subject.update_message({ name: 'test' }) }.to raise_error(Ably::Exceptions::InvalidRequest, /serial is required/) + end + + it 'works when serial is present' do + result = subject.update_message({ name: 'test', serial: serial }) + expect(result).to be_a(Ably::Models::UpdateDeleteResult) + end + end + + context 'does not mutate the original message (#RSL15c)' do + let(:message) { Ably::Models::Message.new(name: 'test', serial: serial) } + + it 'the original message is unchanged' do + original_json = message.as_json.dup + subject.update_message(message) + expect(message.as_json).to eq(original_json) + expect(message.action).to be_nil + end + end + + context 'with query params (#RSL15f)' do + let(:message) { Ably::Models::Message.new(name: 'test', serial: serial) } + + it 'passes params as qs_params' do + expect(client).to receive(:patch).with( + anything, + anything, + { qs_params: { 'key' => 'value' } } + ).and_return(patch_response) + + subject.update_message(message, nil, { 'key' => 'value' }) + end + end + end + + describe '#delete_message (#RSL15)' do + let(:serial) { 'msg-serial-001' } + let(:patch_response) { instance_double('Faraday::Response', status: 200, body: { 'versionSerial' => 'v1-serial' }) } + let(:client) do + instance_double( + 'Ably::PubSub::Http::Client', + encoders: [], + post: instance_double('Faraday::Response', status: 201, body: { 'serials' => ['serial-001'] }), + patch: patch_response, + idempotent_rest_publishing: false, + max_message_size: max_message_size + ) + end + + context 'with a valid message containing serial' do + let(:message) { Ably::Models::Message.new(name: 'test', data: 'hello', serial: serial) } + + it 'sends a PATCH request with action MESSAGE_DELETE' do + expect(client).to receive(:patch).with( + "/channels/#{channel_name}/messages/#{serial}", + hash_including('action' => Ably::Models::Message::ACTION.MessageDelete.to_i), + {} + ).and_return(patch_response) + + subject.delete_message(message) + end + + it 'returns an UpdateDeleteResult' do + result = subject.delete_message(message) + expect(result).to be_a(Ably::Models::UpdateDeleteResult) + expect(result.version_serial).to eql('v1-serial') + end + end + + context 'with an operation parameter' do + let(:message) { Ably::Models::Message.new(name: 'test', serial: serial) } + let(:operation) { { description: 'Removed by moderator' } } + + it 'includes the operation as version in the payload' do + expect(client).to receive(:patch) do |_path, payload, _opts| + expect(payload['version']).to eq({ 'description' => 'Removed by moderator' }) + patch_response + end + + subject.delete_message(message, operation) + end + end + + context 'without serial' do + let(:message) { Ably::Models::Message.new(name: 'test', data: 'hello') } + + it 'raises an InvalidRequest exception' do + expect { subject.delete_message(message) }.to raise_error(Ably::Exceptions::InvalidRequest, /serial is required/) + end + end + + context 'does not mutate the original message' do + let(:message) { Ably::Models::Message.new(name: 'test', serial: serial) } + + it 'the original message is unchanged' do + original_json = message.as_json.dup + subject.delete_message(message) + expect(message.as_json).to eq(original_json) + expect(message.action).to be_nil + end + end + end + + describe '#append_message (#RSL15)' do + let(:serial) { 'msg-serial-001' } + let(:patch_response) { instance_double('Faraday::Response', status: 200, body: { 'versionSerial' => 'v1-serial' }) } + let(:client) do + instance_double( + 'Ably::PubSub::Http::Client', + encoders: [], + post: instance_double('Faraday::Response', status: 201, body: { 'serials' => ['serial-001'] }), + patch: patch_response, + idempotent_rest_publishing: false, + max_message_size: max_message_size + ) + end + + context 'with a valid message containing serial' do + let(:message) { Ably::Models::Message.new(name: 'test', data: ' appended', serial: serial) } + + it 'sends a PATCH request with action MESSAGE_APPEND' do + expect(client).to receive(:patch).with( + "/channels/#{channel_name}/messages/#{serial}", + hash_including('action' => Ably::Models::Message::ACTION.MessageAppend.to_i), + {} + ).and_return(patch_response) + + subject.append_message(message) + end + + it 'returns an UpdateDeleteResult' do + result = subject.append_message(message) + expect(result).to be_a(Ably::Models::UpdateDeleteResult) + expect(result.version_serial).to eql('v1-serial') + end + end + + context 'with an operation parameter' do + let(:message) { Ably::Models::Message.new(name: 'test', serial: serial) } + let(:operation) { { description: 'Added suffix' } } + + it 'includes the operation as version in the payload' do + expect(client).to receive(:patch) do |_path, payload, _opts| + expect(payload['version']).to eq({ 'description' => 'Added suffix' }) + patch_response + end + + subject.append_message(message, operation) + end + end + + context 'without serial' do + let(:message) { Ably::Models::Message.new(name: 'test', data: 'hello') } + + it 'raises an InvalidRequest exception' do + expect { subject.append_message(message) }.to raise_error(Ably::Exceptions::InvalidRequest, /serial is required/) + end + end + + context 'does not mutate the original message' do + let(:message) { Ably::Models::Message.new(name: 'test', serial: serial) } + + it 'the original message is unchanged' do + original_json = message.as_json.dup + subject.append_message(message) + expect(message.as_json).to eq(original_json) + expect(message.action).to be_nil + end + end + end end diff --git a/spec/unit/models/http_paginated_result_spec.rb b/spec/unit/models/http_paginated_result_spec.rb index af29962b8..b65ee80db 100644 --- a/spec/unit/models/http_paginated_result_spec.rb +++ b/spec/unit/models/http_paginated_result_spec.rb @@ -128,40 +128,32 @@ end if defined?(Ably::PubSub::Realtime) - context 'with option async_blocking_operations: true' do - include RSpec::EventMachine - + context 'with option async_blocking_operations: true', :event_machine do subject do paginated_result_class.new(http_response, full_url, paged_client, async_blocking_operations: true) end context '#next' do it 'returns a SafeDeferrable that catches exceptions in callbacks and logs them' do - run_reactor do - expect(subject.next).to be_a(Ably::Util::SafeDeferrable) - stop_reactor - end + expect(subject.next).to be_a(Ably::Util::SafeDeferrable) + stop_reactor end it 'allows a success callback block to be added' do - run_reactor do - subject.next do |paginated_result| - expect(paginated_result).to be_a(Ably::Models::HttpPaginatedResponse) - stop_reactor - end + subject.next do |paginated_result| + expect(paginated_result).to be_a(Ably::Models::HttpPaginatedResponse) + stop_reactor end end end context '#first' do it 'calls the errback callback when first page headers are missing' do - run_reactor do - subject.next do |paginated_result| - deferrable = subject.first - deferrable.errback do |error| - expect(error).to be_a(Ably::Exceptions::PageMissing) - stop_reactor - end + subject.next do |paginated_result| + deferrable = subject.first + deferrable.errback do |error| + expect(error).to be_a(Ably::Exceptions::PageMissing) + stop_reactor end end end diff --git a/spec/unit/models/message_operation_spec.rb b/spec/unit/models/message_operation_spec.rb new file mode 100644 index 000000000..85acc8df5 --- /dev/null +++ b/spec/unit/models/message_operation_spec.rb @@ -0,0 +1,67 @@ +# encoding: utf-8 +require 'spec_helper' + +describe Ably::Models::MessageOperation do + subject { Ably::Models::MessageOperation } + + context '#client_id (#MOP2a)' do + let(:model) { subject.new(client_id: 'user123') } + + it 'returns the client_id' do + expect(model.client_id).to eql('user123') + end + end + + context '#description (#MOP2b)' do + let(:model) { subject.new(description: 'Edited for clarity') } + + it 'returns the description' do + expect(model.description).to eql('Edited for clarity') + end + end + + context '#metadata (#MOP2c)' do + let(:model) { subject.new(metadata: { 'reason' => 'typo' }) } + + it 'returns the metadata hash' do + expect(model.metadata).to eq({ 'reason' => 'typo' }) + end + end + + context 'when empty' do + let(:model) { subject.new({}) } + + it 'returns nil for all fields' do + expect(model.client_id).to be_nil + expect(model.description).to be_nil + expect(model.metadata).to be_nil + end + end + + context 'with camelCase keys from wire' do + let(:model) { subject.new('clientId' => 'user456') } + + it 'converts to snake_case access' do + expect(model.client_id).to eql('user456') + end + end + + context '#attributes' do + let(:model) { subject.new(description: 'test') } + + it 'prevents modification' do + expect { model.attributes[:description] = 'changed' }.to raise_error(/can't modify frozen|FrozenError/) + end + end + + context '#as_json' do + let(:model) { subject.new(client_id: 'user', description: 'edit') } + + it 'returns a hash suitable for JSON serialization' do + json = model.as_json + expect(json).to be_a(Hash) + expect(json['clientId']).to eql('user') + expect(json['description']).to eql('edit') + end + end +end diff --git a/spec/unit/models/message_spec.rb b/spec/unit/models/message_spec.rb index a2939920f..eb21f78f6 100644 --- a/spec/unit/models/message_spec.rb +++ b/spec/unit/models/message_spec.rb @@ -644,4 +644,169 @@ end end end + + context '#action (#TM2j)' do + context 'when action is present' do + let(:model) { subject.new({ action: 1 }) } + + it 'returns the action as an ACTION enum' do + expect(model.action).to eq(Ably::Models::Message::ACTION.MessageUpdate) + end + + it 'can be compared with a symbol' do + expect(model.action).to eq(:message_update) + end + + it 'can be compared with an integer' do + expect(model.action).to eq(1) + end + end + + context 'when action is not present' do + let(:model) { subject.new({}) } + + it 'returns nil' do + expect(model.action).to be_nil + end + end + + context 'ACTION enum values (#TM5)' do + it 'has message_create as 0' do + expect(Ably::Models::Message::ACTION.MessageCreate.to_i).to eq(0) + end + + it 'has message_update as 1' do + expect(Ably::Models::Message::ACTION.MessageUpdate.to_i).to eq(1) + end + + it 'has message_delete as 2' do + expect(Ably::Models::Message::ACTION.MessageDelete.to_i).to eq(2) + end + + it 'has meta as 3' do + expect(Ably::Models::Message::ACTION.Meta.to_i).to eq(3) + end + + it 'has message_summary as 4' do + expect(Ably::Models::Message::ACTION.MessageSummary.to_i).to eq(4) + end + + it 'has message_append as 5' do + expect(Ably::Models::Message::ACTION.MessageAppend.to_i).to eq(5) + end + end + end + + context '#serial (#TM2r)' do + let(:serial_value) { 'msg-serial-001' } + let(:model) { subject.new({ serial: serial_value }) } + + it 'returns the serial attribute' do + expect(model.serial).to eql(serial_value) + end + + context 'when not present' do + let(:model) { subject.new({}) } + + it 'returns nil' do + expect(model.serial).to be_nil + end + end + end + + context '#version (#TM2s)' do + let(:version_data) { { 'serial' => 'v1', 'timestamp' => 1234567890 } } + let(:model) { subject.new({ version: version_data }) } + + it 'returns the version attribute' do + expect(model.version).to eq(version_data) + end + + context 'when not present' do + let(:model) { subject.new({}) } + + it 'returns nil' do + expect(model.version).to be_nil + end + end + end + + context '#created_at' do + let(:time_ms) { (Time.now.to_f * 1000).to_i } + let(:model) { subject.new({ created_at: time_ms }) } + + it 'returns a Time object' do + expect(model.created_at).to be_a(Time) + end + + context 'when not present' do + let(:model) { subject.new({}) } + + it 'returns nil' do + expect(model.created_at).to be_nil + end + end + end + + context '#updated_at' do + let(:time_ms) { (Time.now.to_f * 1000).to_i } + let(:model) { subject.new({ updated_at: time_ms }) } + + it 'returns a Time object' do + expect(model.updated_at).to be_a(Time) + end + + context 'when not present' do + let(:model) { subject.new({}) } + + it 'returns nil' do + expect(model.updated_at).to be_nil + end + end + end + + context '#as_json' do + context 'with action' do + let(:model) { subject.new({ name: 'test', action: 1 }) } + + it 'converts action to integer' do + expect(model.as_json['action']).to eq(1) + end + + it 'includes name' do + expect(model.as_json['name']).to eq('test') + end + end + + context 'without action' do + let(:model) { subject.new({ name: 'test' }) } + + it 'does not include action key' do + expect(model.as_json).not_to have_key('action') + end + end + + context 'with serial and version' do + let(:model) { subject.new({ serial: 'abc', version: { 'description' => 'edit' } }) } + + it 'includes serial' do + expect(model.as_json['serial']).to eq('abc') + end + + it 'includes version' do + expect(model.as_json['version']).to eq({ 'description' => 'edit' }) + end + end + + context 'excludes nil values' do + let(:model) { subject.new({ name: 'test' }) } + + it 'does not include nil attributes' do + json = model.as_json + expect(json).not_to have_key('serial') + expect(json).not_to have_key('version') + expect(json).not_to have_key('encoding') + end + end + end end diff --git a/spec/unit/models/paginated_result_spec.rb b/spec/unit/models/paginated_result_spec.rb index cc0cfd1dc..b15b41b3e 100644 --- a/spec/unit/models/paginated_result_spec.rb +++ b/spec/unit/models/paginated_result_spec.rb @@ -126,40 +126,32 @@ end if defined?(Ably::PubSub::Realtime) - context 'with option async_blocking_operations: true' do - include RSpec::EventMachine - + context 'with option async_blocking_operations: true', :event_machine do subject do paginated_result_class.new(http_response, full_url, paged_client, async_blocking_operations: true) end context '#next' do it 'returns a SafeDeferrable that catches exceptions in callbacks and logs them' do - run_reactor do - expect(subject.next).to be_a(Ably::Util::SafeDeferrable) - stop_reactor - end + expect(subject.next).to be_a(Ably::Util::SafeDeferrable) + stop_reactor end it 'allows a success callback block to be added' do - run_reactor do - subject.next do |paginated_result| - expect(paginated_result).to be_a(Ably::Models::PaginatedResult) - stop_reactor - end + subject.next do |paginated_result| + expect(paginated_result).to be_a(Ably::Models::PaginatedResult) + stop_reactor end end end context '#first' do it 'calls the errback callback when first page headers are missing' do - run_reactor do - subject.next do |paginated_result| - deferrable = subject.first - deferrable.errback do |error| - expect(error).to be_a(Ably::Exceptions::PageMissing) - stop_reactor - end + subject.next do |paginated_result| + deferrable = subject.first + deferrable.errback do |error| + expect(error).to be_a(Ably::Exceptions::PageMissing) + stop_reactor end end end diff --git a/spec/unit/models/protocol_message_spec.rb b/spec/unit/models/protocol_message_spec.rb index 32c74cbc3..67ab2eabf 100644 --- a/spec/unit/models/protocol_message_spec.rb +++ b/spec/unit/models/protocol_message_spec.rb @@ -215,6 +215,46 @@ def new_protocol_message(options) end end + context '#res (#TR4s)' do + context 'when present' do + let(:res_data) { [{ 'serials' => ['serial-001', 'serial-002'] }] } + let(:protocol_message) { new_protocol_message(res: res_data) } + + it 'returns the res array' do + expect(protocol_message.res).to be_a(Array) + expect(protocol_message.res.length).to eql(1) + end + + it 'contains publish result entries with serials' do + entry = protocol_message.res[0] + expect(entry['serials']).to eq(['serial-001', 'serial-002']) + end + end + + context 'when absent' do + let(:protocol_message) { new_protocol_message({}) } + + it 'returns nil' do + expect(protocol_message.res).to be_nil + end + end + + context 'with multiple entries' do + let(:res_data) do + [ + { 'serials' => ['serial-a'] }, + { 'serials' => ['serial-b', 'serial-c'] } + ] + end + let(:protocol_message) { new_protocol_message(res: res_data) } + + it 'returns all entries' do + expect(protocol_message.res.length).to eql(2) + expect(protocol_message.res[1]['serials']).to eq(['serial-b', 'serial-c']) + end + end + end + context '#params (#RTL4k1)' do let(:params) do { foo: :bar } diff --git a/spec/unit/models/publish_result_spec.rb b/spec/unit/models/publish_result_spec.rb new file mode 100644 index 000000000..5a724c837 --- /dev/null +++ b/spec/unit/models/publish_result_spec.rb @@ -0,0 +1,68 @@ +# encoding: utf-8 +require 'spec_helper' + +describe Ably::Models::PublishResult do + subject { Ably::Models::PublishResult } + + context '#serials (#RSL1n)' do + context 'when present' do + let(:model) { subject.new(serials: ['serial-001', 'serial-002']) } + + it 'returns the serials array' do + expect(model.serials).to eql(['serial-001', 'serial-002']) + end + end + + context 'with nullable entries' do + let(:model) { subject.new(serials: ['serial-001', nil, 'serial-003']) } + + it 'preserves nil entries' do + expect(model.serials).to eql(['serial-001', nil, 'serial-003']) + end + end + + context 'when empty array' do + let(:model) { subject.new(serials: []) } + + it 'returns empty array' do + expect(model.serials).to eql([]) + end + end + + context 'when nil' do + let(:model) { subject.new(serials: nil) } + + it 'returns empty array' do + expect(model.serials).to eql([]) + end + end + + context 'when not provided' do + let(:model) { subject.new({}) } + + it 'returns empty array' do + expect(model.serials).to eql([]) + end + end + end + + context '#attributes' do + let(:model) { subject.new(serials: ['s1']) } + + it 'returns the underlying attributes' do + expect(model.attributes).to be_a(Ably::Models::IdiomaticRubyWrapper) + end + + it 'prevents modification' do + expect { model.attributes[:serials] = ['changed'] }.to raise_error(/can't modify frozen|FrozenError/) + end + end + + context 'truthiness' do + let(:model) { subject.new(serials: []) } + + it 'is truthy for backward compatibility with boolean publish returns' do + expect(model).to be_truthy + end + end +end diff --git a/spec/unit/models/stats_spec.rb b/spec/unit/models/stats_spec.rb index f2304163a..657dc2cb8 100644 --- a/spec/unit/models/stats_spec.rb +++ b/spec/unit/models/stats_spec.rb @@ -7,199 +7,64 @@ subject { Ably::Models::Stats } - %w(all persisted).each do |attribute| - context "##{attribute} stats" do - let(:data) do - { attribute.to_sym => { messages: { count: 5 }, all: { data: 10 } } } - end - subject { Ably::Models::Stats.new(data.merge(interval_id: '2004-02')).public_send(attribute) } - - it 'returns a MessageTypes object' do - expect(subject).to be_a(Ably::Models::Stats::MessageTypes) - end - - it 'returns value for message counts' do - expect(subject.messages.count).to eql(5) - end - - it 'returns value for all data transferred' do - expect(subject.all.data).to eql(10) - end - - it 'returns zero for empty values' do - expect(subject.presence.count).to eql(0) - end - - it 'raises an exception for unknown attributes' do - expect { subject.unknown }.to raise_error NoMethodError - end - - %w(all presence messages).each do |type| - context "##{type}" do - it 'is a MessageCount object' do - expect(subject.public_send(type)).to be_a(Ably::Models::Stats::MessageCount) - end - end - end + describe '#interval_id' do + it 'returns the interval ID string' do + stat = subject.new(interval_id: '2024-02-03:15:05', unit: 'minute', entries: {}, schema: 'https://schemas.ably.com/json/app-stats-0.0.5.json') + expect(stat.interval_id).to eql('2024-02-03:15:05') end end - %w(inbound outbound).each do |direction| - context "##{direction} stats" do - let(:data) do - { - direction.to_sym => { - realtime: { messages: { count: 5 }, presence: { data: 10 } }, - all: { messages: { count: 25 }, presence: { data: 210 } } - } - } - end - subject { Ably::Models::Stats.new(data.merge(interval_id: '2004-02')).public_send(direction) } - - it 'returns a MessageTraffic object' do - expect(subject).to be_a(Ably::Models::Stats::MessageTraffic) - end - - it 'returns value for realtime message counts' do - expect(subject.realtime.messages.count).to eql(5) - end - - it 'returns value for all presence data' do - expect(subject.all.presence.data).to eql(210) - end - - it 'raises an exception for unknown attributes' do - expect { subject.unknown }.to raise_error NoMethodError - end - - %w(realtime rest webhook all).each do |type| - context "##{type}" do - it 'is a MessageTypes object' do - expect(subject.public_send(type)).to be_a(Ably::Models::Stats::MessageTypes) - end - end - end + describe '#interval_time' do + it 'returns a Time object representing the start of the interval' do + stat = subject.new(interval_id: '2004-02-01:05:06', unit: 'minute', entries: {}, schema: 'https://schemas.ably.com/json/app-stats-0.0.5.json') + expect(stat.interval_time.to_i).to eql(Time.new(2004, 02, 01, 05, 06, 00, '+00:00').to_i) end end - context '#connections stats' do - let(:data) do - { connections: { tls: { opened: 5 }, all: { peak: 10 } } } - end - subject { Ably::Models::Stats.new(data.merge(interval_id: '2004-02')).connections } - - it 'returns a ConnectionTypes object' do - expect(subject).to be_a(Ably::Models::Stats::ConnectionTypes) - end - - it 'returns value for tls opened counts' do - expect(subject.tls.opened).to eql(5) - end - - it 'returns value for all peak connections' do - expect(subject.all.peak).to eql(10) - end - - it 'returns zero for empty values' do - expect(subject.all.refused).to eql(0) - end - - it 'raises an exception for unknown attributes' do - expect { subject.unknown }.to raise_error NoMethodError - end - - %w(tls plain all).each do |type| - context "##{type}" do - it 'is a ResourceCount object' do - expect(subject.public_send(type)).to be_a(Ably::Models::Stats::ResourceCount) - end - end + describe '#unit' do + it 'returns the unit from the JSON response' do + stat = subject.new(interval_id: '2024-02', unit: 'month', entries: {}, schema: 'https://schemas.ably.com/json/app-stats-0.0.5.json') + expect(stat.unit).to eql('month') end end - context '#channels stats' do - let(:data) do - { channels: { opened: 5, peak: 10 } } - end - subject { Ably::Models::Stats.new(data.merge(interval_id: '2004-02')).channels } - - it 'returns a ResourceCount object' do - expect(subject).to be_a(Ably::Models::Stats::ResourceCount) + describe '#entries' do + it 'returns the entries hash' do + entries = { 'messages.all.all.count' => 100, 'channels.peak' => 50 } + stat = subject.new(interval_id: '2024-02', unit: 'month', entries: entries, schema: 'https://schemas.ably.com/json/app-stats-0.0.5.json') + expect(stat.entries['messages.all.all.count']).to eql(100) + expect(stat.entries['channels.peak']).to eql(50) end - it 'returns value for opened counts' do - expect(subject.opened).to eql(5) - end - - it 'returns value for peak channels' do - expect(subject.peak).to eql(10) - end - - it 'returns zero for empty values' do - expect(subject.refused).to eql(0) - end - - it 'raises an exception for unknown attributes' do - expect { subject.unknown }.to raise_error NoMethodError - end - - %w(opened peak mean min refused).each do |type| - context "##{type}" do - it 'is a Integer object' do - expect(subject.public_send(type)).to be_a(Integer) - end - end + it 'returns an empty hash when entries is not present' do + stat = subject.new(interval_id: '2024-02', unit: 'month', schema: 'https://schemas.ably.com/json/app-stats-0.0.5.json') + expect(stat.entries).to eql({}) end end - %w(api_requests token_requests).each do |request_type| - context "##{request_type} stats" do - let(:data) do - { - request_type.to_sym => { succeeded: 5, failed: 10 } - } - end - subject { Ably::Models::Stats.new(data.merge(interval_id: '2004-02')).public_send(request_type) } - - it 'returns a RequestCount object' do - expect(subject).to be_a(Ably::Models::Stats::RequestCount) - end - - it 'returns value for succeeded' do - expect(subject.succeeded).to eql(5) - end - - it 'returns value for failed' do - expect(subject.failed).to eql(10) - end - - it 'raises an exception for unknown attributes' do - expect { subject.unknown }.to raise_error NoMethodError - end + describe '#in_progress' do + it 'returns the in_progress string when present' do + stat = subject.new(interval_id: '2024-02', unit: 'month', entries: {}, schema: 'https://schemas.ably.com/json/app-stats-0.0.5.json', inProgress: '2024-02-15:10:30') + expect(stat.in_progress).to eql('2024-02-15:10:30') + end - %w(succeeded failed refused).each do |type| - context "##{type}" do - it 'is a Integer object' do - expect(subject.public_send(type)).to be_a(Integer) - end - end - end + it 'returns nil when not present' do + stat = subject.new(interval_id: '2024-02', unit: 'month', entries: {}, schema: 'https://schemas.ably.com/json/app-stats-0.0.5.json') + expect(stat.in_progress).to be_nil end end - describe '#interval_granularity' do - subject { Ably::Models::Stats.new(interval_id: '2004-02') } - - it 'returns the granularity of the interval_id' do - expect(subject.interval_granularity).to eq(:month) + describe '#schema' do + it 'returns the schema URI' do + stat = subject.new(interval_id: '2024-02', unit: 'month', entries: {}, schema: 'https://schemas.ably.com/json/app-stats-0.0.5.json') + expect(stat.schema).to eql('https://schemas.ably.com/json/app-stats-0.0.5.json') end end - describe '#interval_time' do - subject { Ably::Models::Stats.new(interval_id: '2004-02-01:05:06') } - - it 'returns a Time object representing the start of the interval' do - expect(subject.interval_time.to_i).to eql(Time.new(2004, 02, 01, 05, 06, 00, '+00:00').to_i) + describe '#app_id' do + it 'returns the application ID' do + stat = subject.new(interval_id: '2024-02', unit: 'month', entries: {}, schema: 'https://schemas.ably.com/json/app-stats-0.0.5.json', appId: 'app123') + expect(stat.app_id).to eql('app123') end end diff --git a/spec/unit/models/update_delete_result_spec.rb b/spec/unit/models/update_delete_result_spec.rb new file mode 100644 index 000000000..cfd13dbb1 --- /dev/null +++ b/spec/unit/models/update_delete_result_spec.rb @@ -0,0 +1,52 @@ +# encoding: utf-8 +require 'spec_helper' + +describe Ably::Models::UpdateDeleteResult do + subject { Ably::Models::UpdateDeleteResult } + + context '#version_serial (#UDR2a)' do + context 'when present' do + let(:model) { subject.new(version_serial: 'v1-serial') } + + it 'returns the version serial' do + expect(model.version_serial).to eql('v1-serial') + end + end + + context 'when nil' do + let(:model) { subject.new(version_serial: nil) } + + it 'returns nil' do + expect(model.version_serial).to be_nil + end + end + + context 'when not provided' do + let(:model) { subject.new({}) } + + it 'returns nil' do + expect(model.version_serial).to be_nil + end + end + end + + context 'with camelCase keys from wire' do + let(:model) { subject.new('versionSerial' => 'v1-serial') } + + it 'converts to snake_case access' do + expect(model.version_serial).to eql('v1-serial') + end + end + + context '#attributes' do + let(:model) { subject.new(version_serial: 'v1') } + + it 'returns the underlying attributes' do + expect(model.attributes).to be_a(Ably::Models::IdiomaticRubyWrapper) + end + + it 'prevents modification' do + expect { model.attributes[:version_serial] = 'changed' }.to raise_error(/can't modify frozen|FrozenError/) + end + end +end diff --git a/spec/unit/realtime/incoming_message_dispatcher_spec.rb b/spec/unit/realtime/incoming_message_dispatcher_spec.rb index 5d60d0fd3..2cd23e400 100644 --- a/spec/unit/realtime/incoming_message_dispatcher_spec.rb +++ b/spec/unit/realtime/incoming_message_dispatcher_spec.rb @@ -33,4 +33,206 @@ msgbus.publish :protocol_message, Ably::Models::ProtocolMessage.new(:action => :attached, channel: 'unknown') end end + + context '#ack_messages' do + let(:dispatcher) { subject } + let(:logger) { instance_double('Logger') } + + before do + allow(dispatcher).to receive(:logger).and_return(logger) + allow(logger).to receive(:debug) + end + + context 'with a regular message (no action)' do + let(:message) do + msg = Ably::Models::Message.new(name: 'test', data: 'hello') + # SafeDeferrable is included when Realtime is loaded + msg + end + + it 'succeeds the message with itself' do + succeeded = false + message.callback do |result| + expect(result).to be(message) + succeeded = true + end + + dispatcher.send(:ack_messages, [message]) + expect(succeeded).to be(true) + end + end + + context 'with a MESSAGE_UPDATE action and publish_result' do + let(:message) do + Ably::Models::Message.new( + name: 'test', + data: 'updated', + serial: 'msg-serial-001', + action: Ably::Models::Message::ACTION.MessageUpdate.to_i + ) + end + let(:publish_result) { { 'serials' => ['version-serial-abc'] } } + + it 'succeeds with an UpdateDeleteResult' do + succeeded = false + message.callback do |result| + expect(result).to be_a(Ably::Models::UpdateDeleteResult) + expect(result.version_serial).to eql('version-serial-abc') + succeeded = true + end + + dispatcher.send(:ack_messages, [message], publish_result) + expect(succeeded).to be(true) + end + end + + context 'with a MESSAGE_UPDATE action but no publish_result' do + let(:message) do + Ably::Models::Message.new( + name: 'test', + serial: 'msg-serial-001', + action: Ably::Models::Message::ACTION.MessageUpdate.to_i + ) + end + + it 'succeeds the message with itself (fallback)' do + succeeded = false + message.callback do |result| + expect(result).to be(message) + succeeded = true + end + + dispatcher.send(:ack_messages, [message]) + expect(succeeded).to be(true) + end + end + + context 'with multiple update messages and a publish_result with multiple serials' do + let(:message1) do + Ably::Models::Message.new( + name: 'test1', + serial: 'serial-1', + action: Ably::Models::Message::ACTION.MessageUpdate.to_i + ) + end + let(:message2) do + Ably::Models::Message.new( + name: 'test2', + serial: 'serial-2', + action: Ably::Models::Message::ACTION.MessageUpdate.to_i + ) + end + let(:publish_result) { { 'serials' => ['vs-aaa', 'vs-bbb'] } } + + it 'assigns the correct version serial to each message by index' do + results = [] + message1.callback { |r| results << r } + message2.callback { |r| results << r } + + dispatcher.send(:ack_messages, [message1, message2], publish_result) + + expect(results.length).to eql(2) + expect(results[0]).to be_a(Ably::Models::UpdateDeleteResult) + expect(results[0].version_serial).to eql('vs-aaa') + expect(results[1]).to be_a(Ably::Models::UpdateDeleteResult) + expect(results[1].version_serial).to eql('vs-bbb') + end + end + + context 'with symbol-keyed publish_result' do + let(:message) do + Ably::Models::Message.new( + name: 'test', + serial: 'msg-serial-001', + action: Ably::Models::Message::ACTION.MessageUpdate.to_i + ) + end + let(:publish_result) { { serials: ['vs-sym'] } } + + it 'handles symbol keys for serials' do + succeeded = false + message.callback do |result| + expect(result).to be_a(Ably::Models::UpdateDeleteResult) + expect(result.version_serial).to eql('vs-sym') + succeeded = true + end + + dispatcher.send(:ack_messages, [message], publish_result) + expect(succeeded).to be(true) + end + end + + context 'with a regular message and publish_result (PublishResult)' do + let(:message) do + Ably::Models::Message.new(name: 'test', data: 'hello') + end + let(:publish_result) { { 'serials' => ['pub-serial-001'] } } + + it 'succeeds with a PublishResult' do + succeeded = false + message.callback do |result| + expect(result).to be_a(Ably::Models::PublishResult) + expect(result.serials).to eql(['pub-serial-001']) + succeeded = true + end + + dispatcher.send(:ack_messages, [message], publish_result) + expect(succeeded).to be(true) + end + end + + context 'with multiple regular messages and publish_result' do + let(:message1) { Ably::Models::Message.new(name: 'test1', data: 'hello') } + let(:message2) { Ably::Models::Message.new(name: 'test2', data: 'world') } + let(:publish_result) { { 'serials' => ['ser-aaa', 'ser-bbb'] } } + + it 'assigns the correct serial to each message by index' do + results = [] + message1.callback { |r| results << r } + message2.callback { |r| results << r } + + dispatcher.send(:ack_messages, [message1, message2], publish_result) + + expect(results.length).to eql(2) + expect(results[0]).to be_a(Ably::Models::PublishResult) + expect(results[0].serials).to eql(['ser-aaa']) + expect(results[1]).to be_a(Ably::Models::PublishResult) + expect(results[1].serials).to eql(['ser-bbb']) + end + end + + context 'with a regular message and symbol-keyed publish_result' do + let(:message) { Ably::Models::Message.new(name: 'test', data: 'hello') } + let(:publish_result) { { serials: ['sym-serial'] } } + + it 'handles symbol keys for serials' do + succeeded = false + message.callback do |result| + expect(result).to be_a(Ably::Models::PublishResult) + expect(result.serials).to eql(['sym-serial']) + succeeded = true + end + + dispatcher.send(:ack_messages, [message], publish_result) + expect(succeeded).to be(true) + end + end + + context 'with presence messages' do + let(:presence_message) do + Ably::Models::PresenceMessage.new(action: :enter, client_id: 'user1') + end + + it 'succeeds with the message itself (no publish_result handling)' do + succeeded = false + presence_message.callback do |result| + expect(result).to be(presence_message) + succeeded = true + end + + dispatcher.send(:ack_messages, [presence_message]) + expect(succeeded).to be(true) + end + end + end end