lib/submodules/ably-ruby/lib/ably/realtime/presence.rb in ably-rest-0.9.3 vs lib/submodules/ably-ruby/lib/ably/realtime/presence.rb in ably-rest-1.0.0

- old
+ new

@@ -11,25 +11,19 @@ STATE = ruby_enum('STATE', :initialized, :entering, :entered, :leaving, - :left, - :failed + :left ) include Ably::Modules::StateEmitter include Ably::Modules::UsesStateMachine # {Ably::Realtime::Channel} this Presence object is associated with # @return [Ably::Realtime::Channel] attr_reader :channel - # A unique identifier for this channel client based on their connection, disambiguating situations - # where a given client_id is present on multiple connections simultaneously. - # @return [String] - attr_reader :connection_id - # The client_id for the member present on this channel # @return [String] attr_reader :client_id # The data for the member present on this channel @@ -70,25 +64,28 @@ ensure_supported_payload data @data = data return deferrable_succeed(deferrable, &success_block) if state == STATE.Entered - ensure_presence_publishable_on_connection + requirements_failed_deferrable = ensure_presence_publishable_on_connection_deferrable + return requirements_failed_deferrable if requirements_failed_deferrable + ensure_channel_attached(deferrable) do if entering? once_or_if(STATE.Entered, else: proc { |args| deferrable_fail deferrable, *args }) do deferrable_succeed deferrable, &success_block end else + current_state = state change_state STATE.Entering send_protocol_message_and_transition_state_to( Ably::Models::PresenceMessage::ACTION.Enter, deferrable: deferrable, target_state: STATE.Entered, data: data, client_id: client_id, - failed_state: STATE.Failed, + failed_state: current_state, # return to current state if enter fails &success_block ) end end end @@ -123,31 +120,33 @@ # def leave(data = nil, &success_block) deferrable = create_deferrable ensure_supported_payload data - raise Ably::Exceptions::Standard.new('Unable to leave presence channel that is not entered', 400, 91002) unless able_to_leave? @data = data return deferrable_succeed(deferrable, &success_block) if state == STATE.Left - ensure_presence_publishable_on_connection + requirements_failed_deferrable = ensure_presence_publishable_on_connection_deferrable + return requirements_failed_deferrable if requirements_failed_deferrable + ensure_channel_attached(deferrable) do if leaving? once_or_if(STATE.Left, else: proc { |error|deferrable_fail deferrable, *args }) do deferrable_succeed deferrable, &success_block end else + current_state = state change_state STATE.Leaving send_protocol_message_and_transition_state_to( Ably::Models::PresenceMessage::ACTION.Leave, deferrable: deferrable, target_state: STATE.Left, data: data, client_id: client_id, - failed_state: STATE.Failed, + failed_state: current_state, # return to current state if leave fails &success_block ) end end end @@ -181,11 +180,13 @@ ensure_supported_payload data @data = data - ensure_presence_publishable_on_connection + requirements_failed_deferrable = ensure_presence_publishable_on_connection_deferrable + return requirements_failed_deferrable if requirements_failed_deferrable + ensure_channel_attached(deferrable) do send_protocol_message_and_transition_state_to( Ably::Models::PresenceMessage::ACTION.Update, deferrable: deferrable, target_state: STATE.Entered, @@ -212,25 +213,39 @@ ensure_supported_payload data send_presence_action_for_client(Ably::Models::PresenceMessage::ACTION.Update, client_id, data, &success_block) end - # Get the presence state for this Channel. + # Get the presence members for this Channel. # # @param (see Ably::Realtime::Presence::MembersMap#get) # @option options (see Ably::Realtime::Presence::MembersMap#get) # @yield (see Ably::Realtime::Presence::MembersMap#get) # @return (see Ably::Realtime::Presence::MembersMap#get) # def get(options = {}, &block) deferrable = create_deferrable - ensure_channel_attached(deferrable) do + # #RTP11d Don't return PresenceMap when wait for sync is true + # if the map is stale + wait_for_sync = options.fetch(:wait_for_sync, true) + if wait_for_sync && channel.suspended? + EventMachine.next_tick do + deferrable.fail Ably::Exceptions::InvalidState.new( + 'Presence state is out of sync as channel is SUSPENDED. Presence#get on a SUSPENDED channel is only supported with option wait_for_sync: false', + nil, + 91005 + ) + end + return deferrable + end + + ensure_channel_attached(deferrable, allow_suspended: true) do members.get(options).tap do |members_map_deferrable| - members_map_deferrable.callback do |*args| - safe_yield(block, *args) if block_given? - deferrable.succeed(*args) + members_map_deferrable.callback do |members| + safe_yield(block, members) if block_given? + deferrable.succeed(members) end members_map_deferrable.errback do |*args| deferrable.fail(*args) end end @@ -244,13 +259,12 @@ # @yield [Ably::Models::PresenceMessage] For each presence state change event, the block is called # # @return [void] # def subscribe(*actions, &callback) - ensure_channel_attached do - super - end + implicit_attach + super end # Unsubscribe the matching block for presence events on the associated Channel. # If a block is not provided, all subscriptions will be unsubscribed # @@ -277,11 +291,14 @@ # # @return [Ably::Util::SafeDeferrable] # def history(options = {}, &callback) if options.delete(:until_attach) - raise ArgumentError, 'option :until_attach cannot be specified if the channel is not attached' unless channel.attached? + unless channel.attached? + error = Ably::Exceptions::InvalidRequest.new('option :until_attach is invalid as the channel is not attached') + return Ably::Util::SafeDeferrable.new_and_fail_immediately(logger, error) + end options[:from_serial] = channel.attached_serial end async_wrap(callback) do rest_presence.history(options.merge(async_blocking_operations: true)) @@ -295,17 +312,10 @@ @__incoming_msgbus__ ||= Ably::Util::PubSub.new( coerce_into: Proc.new { |event| Ably::Models::ProtocolMessage::ACTION(event) } ) end - # Configure the connection ID for this presence channel. - # Typically configured only once when a user first enters a presence channel. - # @api private - def set_connection_id(new_connection_id) - @connection_id = new_connection_id - end - # Used by {Ably::Modules::StateEmitter} to debug action changes # @api private def logger client.logger end @@ -314,14 +324,10 @@ def sync_complete? members.sync_complete? end private - def able_to_leave? - entering? || entered? - end - # @return [Ably::Models::PresenceMessage] presence message is returned allowing callbacks to be added def send_presence_protocol_message(presence_action, client_id, data) presence_message = create_presence_message(presence_action, client_id, data) unless presence_message.client_id raise Ably::Exceptions::Standard.new('Unable to enter create presence message without a client_id', 400, 91000) @@ -346,39 +352,44 @@ } Ably::Models::PresenceMessage.new(model, logger: logger).tap do |presence_message| presence_message.encode(client.encoders, channel.options) do |encode_error, error_message| client.logger.error error_message - emit :error, encode_error end end end - def ensure_presence_publishable_on_connection + def ensure_presence_publishable_on_connection_deferrable if !connection.can_publish_messages? - raise Ably::Exceptions::MessageQueueingDisabled.new("Message cannot be published. Client is configured to disallow queueing of messages and connection is currently #{connection.state}") + error = Ably::Exceptions::MessageQueueingDisabled.new("Presence event cannot be published as they cannot be queued when the connection is #{connection.state}") + Ably::Util::SafeDeferrable.new_and_fail_immediately(logger, error) end end - def ensure_channel_attached(deferrable = nil) + def ensure_channel_attached(deferrable = nil, options = {}) if channel.attached? yield + elsif options[:allow_suspended] && channel.suspended? + yield else - attach_channel_then { yield } + attach_channel_then(deferrable) { yield } end deferrable end def ensure_supported_client_id(check_client_id) unless check_client_id - raise Ably::Exceptions::IncompatibleClientId.new('Unable to enter/update/leave presence channel without a client_id', 400, 40012) + raise Ably::Exceptions::IncompatibleClientId.new('Unable to enter/update/leave presence channel without a client_id') end if check_client_id == '*' - raise Ably::Exceptions::IncompatibleClientId.new('Unable to enter/update/leave presence channel with the reserved wildcard client_id', 400, 40012) + raise Ably::Exceptions::IncompatibleClientId.new('Unable to enter/update/leave presence channel with the reserved wildcard client_id') end + unless check_client_id.kind_of?(String) + raise Ably::Exceptions::IncompatibleClientId.new('Unable to enter/update/leave with a non String client_id value') + end unless client.auth.can_assume_client_id?(check_client_id) - raise Ably::Exceptions::IncompatibleClientId.new("Cannot enter with provided client_id '#{check_client_id}' as it is incompatible with the current configured client_id '#{client.client_id}'", 400, 40012) + raise Ably::Exceptions::IncompatibleClientId.new("Cannot enter with provided client_id '#{check_client_id}' as it is incompatible with the current configured client_id '#{client.client_id}'") end end def send_protocol_message_and_transition_state_to(action, options = {}, &success_block) deferrable = options.fetch(:deferrable) { raise ArgumentError, 'option :deferrable is required' } @@ -410,27 +421,38 @@ EventMachine.next_tick { deferrable.fail(*args) } # allow errback to be added to the returned Deferrable deferrable end def send_presence_action_for_client(action, client_id, data, &success_block) - ensure_presence_publishable_on_connection + requirements_failed_deferrable = ensure_presence_publishable_on_connection_deferrable + return requirements_failed_deferrable if requirements_failed_deferrable deferrable = create_deferrable ensure_channel_attached(deferrable) do send_presence_protocol_message(action, client_id, data).tap do |protocol_message| protocol_message.callback { |message| deferrable_succeed deferrable, &success_block } protocol_message.errback { |error| deferrable_fail deferrable, error } end end end - def attach_channel_then + def attach_channel_then(deferrable) if channel.detached? || channel.failed? - raise Ably::Exceptions::InvalidStateChange.new("Operation is not allowed when channel is in #{channel.state}", 400, 91001) + deferrable.fail Ably::Exceptions::InvalidState.new("Operation is not allowed when channel is in #{channel.state}", 400, 91001) else - channel.unsafe_once(Channel::STATE.Attached) { yield } + channel.unsafe_once(:attached, :detached, :failed) do |channel_state_change| + if channel_state_change.current == :attached + yield + else + deferrable.fail Ably::Exceptions::InvalidState.new("Operation failed as channel transitioned to #{channel_state_change.current}", 400, 91001) + end + end channel.attach end + end + + def implicit_attach + channel.attach if channel.initialized? end def client channel.client end