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.

Parameters:



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

def initialize(session, options)
  @mutex = Mutex.new

  @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]

  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.

Returns:

  • (Boolean) 


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.

Returns:

  • (String, nil) 


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.

Returns:

  • (Array) 


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.

Returns:

  • (Array) 


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.

Returns:

  • (Float, :unfiltered) 


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.

Returns:

  • (:command, :distinct, :merge, :raw) 


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

def mode
  @mode
end

#selector ⇒ String? (readonly)

The selector for table items.

Returns:

  • (String, nil) 


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.

Returns:



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.



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

def clear_callbacks
  @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.



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

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

#id ⇒ Fixnum

Returns this subscription’s unique identification number.

Returns:

  • (Fixnum) 


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

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.

Parameters:

  • item_name (String) —

    The name of the item to return the current data for.

Returns:

  • (Hash, Array, nil) —

    A copy of the item data.

Raises:

  • (ArgumentError) 


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

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

  @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.

Parameters:

  • callback (Proc) —

    The callback that is to be run when new data arrives.



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

def on_data(&callback)
  @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|`.

Parameters:

  • callback (Proc) —

    The callback that is to be run when an overflow is reported for this subscription.



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

def on_end_of_snapshot(&callback)
  @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|`.

Parameters:

  • callback (Proc) —

    The callback that is to be run when an overflow is reported for this subscription.



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

def on_overflow(&callback)
  @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.

Parameters:

  • line (String) —

    The line of stream data to process.

Returns:

  • (Boolean) —

    Whether the passed line of stream data was processed by this subscription.



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

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.

Parameters:

  • item_name (String) —

    The name of the item to set the data for.

  • item_data (Hash, Array<Hash>) —

    The new data for the item. If #mode is ‘:merge` this must be a hash. If #mode is `:command` then this must be an `Array<Hash>` and each hash entry must have a unique `:key` value.

Raises:

  • (ArgumentError) 


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

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

  @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.

Parameters:

  • options (Hash) (defaults to: {}) —

    The options to start the subscription with.

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`.



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

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.

Parameters:

  • options (Hash) (defaults to: {}) —

    The options to start the subscription with.



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

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.



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

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.



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

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