module Ably module Realtime # The Channel class represents a Channel belonging to this application. # The Channel instance allows messages to be published and # received, and controls the lifecycle of this instance's # attachment to the channel. # # Channels will always be in one of the following states: # # initialized: 0 # attaching: 1 # attached: 2 # detaching: 3 # detached: 4 # failed: 5 # # Note that the states are available as Enum-like constants: # # Channel::STATE.Initialized # Channel::STATE.Attaching # Channel::STATE.Attached # Channel::STATE.Detaching # Channel::STATE.Detached # Channel::STATE.Failed # # Channels emit errors - use +on(:error)+ to subscribe to errors # # @!attribute [r] state # @return {Ably::Realtime::Connection::STATE} channel state # class Channel include Ably::Modules::Conversions include Ably::Modules::EventEmitter include Ably::Modules::EventMachineHelpers include Ably::Modules::AsyncWrapper include Ably::Modules::MessageEmitter extend Ably::Modules::Enum STATE = ruby_enum('STATE', :initialized, :attaching, :attached, :detaching, :detached, :failed ) include Ably::Modules::StateEmitter include Ably::Modules::UsesStateMachine # Max number of messages to bundle in a single ProtocolMessage MAX_PROTOCOL_MESSAGE_BATCH_SIZE = 50 # {Ably::Realtime::Client} associated with this channel # @return [Ably::Realtime::Client] attr_reader :client # Channel name # @return [String] attr_reader :name # Channel options configured for this channel, see {#initialize} for channel_options # @return [Hash] attr_reader :options # When a channel failure occurs this attribute contains the Ably Exception # @return [Ably::Models::ErrorInfo,Ably::Exceptions::BaseAblyException] attr_reader :error_reason # The Channel manager responsible for attaching, detaching and handling failures for this channel # @return [Ably::Realtime::Channel::ChannelManager] # @api private attr_reader :manager # Serial number assigned to this channel when it was attached # @return [Integer] # @api private attr_reader :attached_serial # Initialize a new Channel object # # @param client [Ably::Rest::Client] # @param name [String] The name of the channel # @param channel_options [Hash] Channel options, currently reserved for Encryption options # @option channel_options [Boolean] :encrypted setting this to true for this channel will encrypt & decrypt all messages automatically # @option channel_options [Hash] :cipher_params A hash of options to configure the encryption. *:key* is required, all other options are optional. See {Ably::Util::Crypto#initialize} for a list of +cipher_params+ options # def initialize(client, name, channel_options = {}) ensure_utf_8 :name, name @client = client @name = name @options = channel_options.clone.freeze @queue = [] @state_machine = ChannelStateMachine.new(self) @state = STATE(state_machine.current_state) @manager = ChannelManager.new(self, client.connection) setup_event_handlers setup_presence end # Publish a message on the channel. # # When publishing a message, if the channel is not attached, the channel is implicitly attached # # @param name [String] The event name of the message # @param data [String,ByteArray] payload for the message # @yield [Ably::Models::Message] On success, will call the block with the {Ably::Models::Message} # @return [Ably::Models::Message] Deferrable {Ably::Models::Message} that supports both success (callback) and failure (errback) callbacks # # @example # channel.publish('click', 'body') # # channel.publish('click', 'body') do |message| # puts "#{message.name} event received with #{message.data}" # end # # channel.publish('click', 'body').errback do |message, error| # puts "#{message.name} was not received, error #{error.message}" # end # def publish(name, data, &success_block) ensure_utf_8 :name, name create_message(name, data).tap do |message| message.callback(&success_block) if block_given? queue_message message end end # Subscribe to messages matching providing event name, or all messages if event name not provided. # # When subscribing to messages, if the channel is not attached, the channel is implicitly attached # # @param names [String] The event name of the message to subscribe to if provided. Defaults to all events. # @yield [Ably::Models::Message] For each message received, the block is called # # @return [void] # def subscribe(*names, &callback) attach unless attached? || attaching? super end # Unsubscribe the matching block for messages matching providing event name, or all messages if event name not provided. # If a block is not provided, all subscriptions will be unsubscribed # # @param names [String] The event name of the message to subscribe to if provided. Defaults to all events. # # @return [void] # def unsubscribe(*names, &callback) super end # Attach to this channel, and call the block if provided when attached. # Attaching to a channel is implicit in when a message is published or #subscribe is called, so it is uncommon # to need to call attach explicitly. # # @yield [Ably::Realtime::Channel] Block is called as soon as this channel is in the Attached state # @return [Ably::Util::SafeDeferrable] Deferrable that supports both success (callback) and failure (errback) callback # def attach(&success_block) transition_state_machine :attaching if can_transition_to?(:attaching) deferrable_for_state_change_to(STATE.Attached, &success_block) end # Detach this channel, and call the block if provided when in a Detached or Failed state # # @yield [Ably::Realtime::Channel] Block is called as soon as this channel is in the Detached or Failed state # @return [void] # def detach(&success_block) raise exception_for_state_change_to(:detaching) if failed? || initialized? transition_state_machine :detaching if can_transition_to?(:detaching) deferrable_for_state_change_to(STATE.Detached, &success_block) end # Presence object for this Channel. This controls this client's # presence on the channel and may also be used to obtain presence information # and change events for other members of the channel. # # When accessing presence, if the channel is not attached, the channel is implicitly attached # # @return {Ably::Realtime::Presence} # def presence attach if initialized? @presence end # Return the message history of the channel # # If the channel is attached, you can retrieve messages published on the channel before the # channel was attached with the option until_attach: true. This is useful when a developer # wishes to display historical messages with the guarantee that no messages have been missed since attach. # # @param (see Ably::Rest::Channel#history) # @option options (see Ably::Rest::Channel#history) # @option options [Boolean] :until_attach When true, the history request will be limited only to messages published before this channel was attached. Channel must be attached # # @yield [Ably::Models::PaginatedResource] First {Ably::Models::PaginatedResource page} of {Ably::Models::Message} objects accessible with {Ably::Models::PaginatedResource#items #items}. # # @return [Ably::Util::SafeDeferrable] # def history(options = {}, &callback) if options.delete(:until_attach) raise ArgumentError, 'option :until_attach is invalid as the channel is not attached' unless attached? options[:from_serial] = attached_serial end async_wrap(callback) do rest_channel.history(options.merge(async_blocking_operations: true)) end end # @!attribute [r] __incoming_msgbus__ # @return [Ably::Util::PubSub] Client library internal channel incoming message bus # @api private def __incoming_msgbus__ @__incoming_msgbus__ ||= Ably::Util::PubSub.new( coerce_into: Proc.new { |event| Ably::Models::ProtocolMessage::ACTION(event) } ) end # @api private def set_failed_channel_error_reason(error) @error_reason = error end # @api private def clear_error_reason @error_reason = nil end # @api private def set_attached_serial(serial) @attached_serial = serial end # Used by {Ably::Modules::StateEmitter} to debug state changes # @api private def logger client.logger end private attr_reader :queue def setup_event_handlers __incoming_msgbus__.subscribe(:message) do |message| message.decode self emit_message message.name, message end on(STATE.Attached) do process_queue end end # Queue message and process queue if channel is attached. # If channel is not yet attached, attempt to attach it before the message queue is processed. def queue_message(message) queue << message if attached? process_queue else attach end end def messages_in_queue? !queue.empty? end # Move messages from Channel Queue into Outgoing Connection Queue def process_queue condition = -> { attached? && messages_in_queue? } non_blocking_loop_while(condition) do send_messages_within_protocol_message queue.shift(MAX_PROTOCOL_MESSAGE_BATCH_SIZE) end end def send_messages_within_protocol_message(messages) client.connection.send_protocol_message( action: Ably::Models::ProtocolMessage::ACTION.Message.to_i, channel: name, messages: messages ) end def create_message(name, data) message = { name: name } message.merge!(data: data) unless data.nil? message.merge!(clientId: client.client_id) if client.client_id Ably::Models::Message.new(message, logger: logger).tap do |message| message.encode self end end def rest_channel client.rest_client.channel(name) end def connection client.connection end def setup_presence @presence ||= Presence.new(self) end end end end require 'ably/realtime/channel/channel_manager' require 'ably/realtime/channel/channel_state_machine'