Class: Lightstreamer::Subscription

Inherits:
Object
  • Object
show all
Defined in:
lib/lightstreamer/subscription.rb

Overview

This class manages a subscription that can be used to stream data from a Session. Subscriptions should always be created using Lightstreamer::Session#build_subscription. Subscriptions start receiving data after #start is called, and streaming subscription data can be consumed by registering an asynchronous data callback using #on_data, or by polling using #item_data.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(session, options) ⇒ Subscription

Initializes a new Lightstreamer subscription with the specified options.



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/lightstreamer/subscription.rb', line 58

def initialize(session, options)
  @session = session

  @items = options.fetch(:items)
  @fields = options.fetch(:fields)
  @mode = options.fetch(:mode).to_sym
  @data_adapter = options[:data_adapter]
  @selector = options[:selector]
  @maximum_update_frequency = sanitize_frequency options[:maximum_update_frequency]

  @data_mutex = Mutex.new

  clear_data
  clear_callbacks
end

Instance Attribute Details

#active ⇒ Boolean (readonly)

Whether this subscription is currently started and actively streaming data. See #start and #stop for details.



50
51
52
 
# File 'lib/lightstreamer/subscription.rb', line 50

def active
  @active
end

#data_adapter ⇒ String? (readonly)

The name of the data adapter from the Lightstreamer session’s adapter set to use, or nil to use the default data adapter.



32
33
34
 
# File 'lib/lightstreamer/subscription.rb', line 32

def data_adapter
  @data_adapter
end

#fields ⇒ Array (readonly)

The names of the fields to subscribe to on the items.



20
21
22
 
# File 'lib/lightstreamer/subscription.rb', line 20

def fields
  @fields
end

#items ⇒ Array (readonly)

The names of the items to subscribe to.



15
16
17
 
# File 'lib/lightstreamer/subscription.rb', line 15

def items
  @items
end

#maximum_update_frequency ⇒ Float, :unfiltered

The maximum number of updates this subscription should receive per second. If this is set to zero, which is the default, then there is no limit on the update frequency. If set to :unfiltered then unfiltered streaming will be used for this subscription and it is possible for overflows to occur (see #on_overflow). If #mode is :raw then the maximum update frequency is treated as :unfiltered regardless of its actual value.



45
46
47
 
# File 'lib/lightstreamer/subscription.rb', line 45

def maximum_update_frequency
  @maximum_update_frequency
end

#mode ⇒ :command, ... (readonly)

The operation mode of this subscription. The four supported operation modes are: :command, :distinct, :merge and :raw. See the Lightstreamer documentation for details on the different modes.



26
27
28
 
# File 'lib/lightstreamer/subscription.rb', line 26

def mode
  @mode
end

#selector ⇒ String? (readonly)

The selector for table items.



37
38
39
 
# File 'lib/lightstreamer/subscription.rb', line 37

def selector
  @selector
end

#session ⇒ Session (readonly)

The session that this subscription is associated with.



10
11
12
 
# File 'lib/lightstreamer/subscription.rb', line 10

def session
  @session
end

Instance Method Details

#clear_callbacks ⇒ Object

Removes all #on_data, #on_overflow and #on_end_of_snapshot callbacks present on this subscription.



216
217
218
 
# File 'lib/lightstreamer/subscription.rb', line 216

def clear_callbacks
  @data_mutex.synchronize { @callbacks = { on_data: [], on_overflow: [], on_end_of_snapshot: [] } }
end

#clear_data ⇒ Object

Clears all current data stored for this subscription. New data will continue to be processed as it becomes available.



150
151
152
 
# File 'lib/lightstreamer/subscription.rb', line 150

def clear_data
  @data_mutex.synchronize { @data = (0...items.size).map { SubscriptionItemData.new } }
end

#id ⇒ Fixnum

Returns this subscription’s unique identification number.



79
80
81
 
# File 'lib/lightstreamer/subscription.rb', line 79

def id
  @id ||= ID_GENERATOR.next
end

#item_data(item_name) ⇒ Hash, ...

Returns a copy of the current data of one of this subscription’s items. If #mode is :merge then the returned object will be a hash of the item’s state, if it is :command then an array of row data for the item will be returned, and if it is :distinct or :raw then just the most recent update received for the item will be returned. The return value will be nil if no data for the item has been set or been received.

Raises:

  • (ArgumentError) 


162
163
164
165
166
167
 
# File 'lib/lightstreamer/subscription.rb', line 162

def item_data(item_name)
  index = @items.index item_name
  raise ArgumentError, 'Unknown item' unless index

  @data_mutex.synchronize { @data[index].data && @data[index].data.dup }
end

#on_data(&callback) ⇒ Object

Adds the passed block to the list of callbacks that will be run when new data for this subscription arrives. The block will be called on a worker thread and so the code that is run by the block must be thread-safe. The arguments passed to the block are ‘|subscription, item_name, item_data, new_data|`. If #mode is :distinct or :raw then item_data and new_data will be the same.



189
190
191
 
# File 'lib/lightstreamer/subscription.rb', line 189

def on_data(&callback)
  @data_mutex.synchronize { @callbacks[:on_data] << callback }
end

#on_end_of_snapshot(&callback) ⇒ Object

Adds the passed block to the list of callbacks that will be run when the server reports an end-of-snapshot notification for this subscription. End-of-snapshot notifications are only sent when #mode is :command or :distinct and ‘snapshot: true` was passed to #start. The block will be called on a worker thread and so the code that is run by the block must be thread-safe. The arguments passed to the block are `|subscription, item_name|`.



211
212
213
 
# File 'lib/lightstreamer/subscription.rb', line 211

def on_end_of_snapshot(&callback)
  @data_mutex.synchronize { @callbacks[:on_end_of_snapshot] << callback }
end

#on_overflow(&callback) ⇒ Object

Adds the passed block to the list of callbacks that will be run when the server reports an overflow for this subscription. This is only relevant when this subscription’s #mode is :command or :raw, or if #maximum_update_frequency is :unfiltered. The block will be called on a worker thread and so the code that is run by the block must be thread-safe. The arguments passed to the block are ‘|subscription, item_name, overflow_size|`.



200
201
202
 
# File 'lib/lightstreamer/subscription.rb', line 200

def on_overflow(&callback)
  @data_mutex.synchronize { @callbacks[:on_overflow] << callback }
end

#process_stream_data(line) ⇒ Boolean

Processes a line of stream data if it is relevant to this subscription. This method is thread-safe and is intended to be called by the session’s processing thread.



228
229
230
231
232
 
# File 'lib/lightstreamer/subscription.rb', line 228

def process_stream_data(line)
  return true if process_update_message UpdateMessage.parse(line, id, items, fields)
  return true if process_overflow_message OverflowMessage.parse(line, id, items)
  return true if process_end_of_snapshot_message EndOfSnapshotMessage.parse(line, id, items)
end

#set_item_data(item_name, item_data) ⇒ Object

Sets the current data for the item with the specified name. This is only allowed when #mode is :command or :merge. Raises an exception if the specified item name or item data is invalid.

Raises:

  • (ArgumentError) 


176
177
178
179
180
181
 
# File 'lib/lightstreamer/subscription.rb', line 176

def set_item_data(item_name, item_data)
  index = @items.index item_name
  raise ArgumentError, 'Unknown item' unless index

  @data_mutex.synchronize { @data[index].set_data item_data, mode }
end

#start(options = {}) ⇒ Object

Starts streaming data for this Lightstreamer subscription. If an error occurs then a LightstreamerError subclass will be raised.

Options Hash (options):

  • :silent (Boolean) —

    Whether the subscription should be started in silent mode. In silent mode the subscription is initiated on the server and begins buffering incoming data, however this data will not be sent to the client for processing until #unsilence is called.

  • :snapshot (Boolean, Fixnum) —

    Controls whether the server should send a snapshot of this subscription’s items. The default value is false which means then the server will not send snapshot information. If set to true then the server will send snapshot information if it is available. If this subscription’s #mode is :distinct then :snapshot can also be an integer specifying the number of events the server should send as part of the snapshot. If this latter option is used, or #mode is :command, then any callbacks registered with #on_end_of_snapshot will be called once the snapshot for each item is complete. This option is ignored when #mode is :raw.



98
99
100
101
102
103
 
# File 'lib/lightstreamer/subscription.rb', line 98

def start(options = {})
  return if @active

  session.control_request(*start_control_request_args(options))
  @active = true
end

#start_control_request_args(options = {}) ⇒ Object

Returns the arguments to pass to to Lightstreamer::Session#control_request in order to start this subscription with the given options.



111
112
113
114
115
116
117
118
119
 
# File 'lib/lightstreamer/subscription.rb', line 111

def start_control_request_args(options = {})
  operation = options[:silent] ? :add_silent : :add

  options = { LS_table: id, LS_mode: mode.to_s.upcase, LS_id: items, LS_schema: fields, LS_selector: selector,
              LS_data_adapter: data_adapter, LS_requested_max_frequency: maximum_update_frequency,
              LS_snapshot: options.fetch(:snapshot, false) }

  [operation, options]
end

#stop ⇒ Object

Stops streaming data for this Lightstreamer subscription. If an error occurs then a LightstreamerError subclass will be raised.



130
131
132
133
 
# File 'lib/lightstreamer/subscription.rb', line 130

def stop
  session.control_request :delete, LS_table: id if @active
  @active = false
end

#unsilence ⇒ Object

Unsilences this subscription if it was initially started in silent mode by passing ‘silent: true` to #start. If this subscription was not started in silent mode then this method has no effect. If an error occurs then a LightstreamerError subclass will be raised.



124
125
126
 
# File 'lib/lightstreamer/subscription.rb', line 124

def unsilence
  session.control_request :start, LS_table: id
end