Class: OpenC3::StreamingWebSocketApi

Inherits:
CmdTlmWebSocketApi show all
Defined in:
lib/openc3/script/web_socket_api.rb

Overview

Streaming API WebSocket

Constant Summary

Constants inherited from WebSocketApi

WebSocketApi::DEFAULT_OPTIONS, WebSocketApi::USER_AGENT

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from CmdTlmWebSocketApi

#generate_url

Methods inherited from WebSocketApi

cable_url, #check_protocol_frame, #connect, #connected?, #disconnect, #generate_auth, #parse_message, #read, #read_message, #subscribe, #unsubscribe, #wait_for_subscribed, #write, #write_action, #write_command

Constructor Details

#initialize(**options) ⇒ StreamingWebSocketApi

Returns a new instance of StreamingWebSocketApi.



430
431
432
433
434
435
 
# File 'lib/openc3/script/web_socket_api.rb', line 430

def initialize(**options)
  @identifier = {
    channel: "StreamingChannel"
  }
  super(**options)
end

Class Method Details

.read_all(items: nil, packets: nil, start_time: nil, end_time: nil, scope: nil, timeout: nil) ⇒ Object

Convenience method to read all data until end marker is received. Omitting end_time streams realtime and endlessly: no end marker is ever sent, so a timeout is the only way the collection ends on its own. Warning: DATA IS STORED IN RAM. Do not use this with large queries



493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
 
# File 'lib/openc3/script/web_socket_api.rb', line 493

def self.read_all(items: nil, packets: nil, start_time: nil, end_time: nil, scope: nil, timeout: nil)
  read_all_start_time = Time.now
  data = []
  self.new do |api|
    api.add(items: items, packets: packets, start_time: start_time, end_time: end_time, scope: scope)
    while true
      batch = api.read
      if batch.nil?
        # A bounded query must receive its end marker; a truncated result
        # returned as if complete is worse than an error. A realtime query
        # never gets one, so a close is an ordinary way for it to end.
        raise "WebSocket closed before end marker" if end_time

        return data
      end
      # An empty batch is the explicit end marker sent after a historical
      # query is complete.
      if batch.empty?
        return data
      else
        data.concat(batch)
      end
      if timeout
        if (Time.now - read_all_start_time) > timeout
          return data
        end
      end
    end
  end
end

Instance Method Details

#add(items: nil, packets: nil, start_time: nil, end_time: nil, scope: nil) ⇒ Object

Request to add data to the stream

arguments: scope: scope name start_time: 64-bit nanoseconds from unix epoch - If not present then realtime end_time: 64-bit nanoseconds from unix epoch - If not present stream forever items: [ [ MODE__CMDORTLM__TARGET__PACKET__ITEM__VALUETYPE__REDUCEDTYPE, item_key] ] MODE - RAW, DECOM, REDUCED_MINUTE, REDUCED_HOUR, or REDUCED_DAY CMDORTLM - CMD or TLM TARGET - Target name PACKET - Packet name ITEM - Item Name VALUETYPE - RAW, CONVERTED, FORMATTED REDUCEDTYPE - MIN, MAX, AVG, STDDEV (only for reduced modes) item_key is an optional shortened name to return the data as packets: [ MODE__CMDORTLM__TARGET__PACKET__VALUETYPE ] MODE - RAW, DECOM, REDUCED_MINUTE, REDUCED_HOUR, or REDUCED_DAY CMDORTLM - CMD or TLM TARGET - Target name PACKET - Packet name VALUETYPE - RAW, CONVERTED, FORMATTED, or PURE (pure means all types as stored in log)



459
460
461
462
463
464
 
# File 'lib/openc3/script/web_socket_api.rb', line 459

def add(items: nil, packets: nil, start_time: nil, end_time: nil, scope: nil)
  times = {}
  times['start_time'] = to_nsec(start_time) if start_time
  times['end_time'] = to_nsec(end_time) if end_time
  stream_action('add', items: items, packets: packets, scope: scope, extra: times)
end

#remove(items: nil, packets: nil, scope: nil) ⇒ Object

Request to remove data from the stream

arguments: scope: scope name items: [ [ MODE__CMDORTLM__TARGET__PACKET__ITEM__VALUETYPE__REDUCEDTYPE] ] MODE - RAW, DECOM, REDUCED_MINUTE, REDUCED_HOUR, or REDUCED_DAY CMDORTLM - CMD or TLM TARGET - Target name PACKET - Packet name ITEM - Item Name VALUETYPE - RAW, CONVERTED, FORMATTED REDUCEDTYPE - MIN, MAX, AVG, STDDEV (only for reduced modes) packets: [ MODE__CMDORTLM__TARGET__PACKET__VALUETYPE ] MODE - RAW, DECOM, REDUCED_MINUTE, REDUCED_HOUR, or REDUCED_DAY CMDORTLM - CMD or TLM TARGET - Target name PACKET - Packet name VALUETYPE - RAW, CONVERTED, FORMATTED, or PURE (pure means all types as stored in log)



485
486
487
 
# File 'lib/openc3/script/web_socket_api.rb', line 485

def remove(items: nil, packets: nil, scope: nil)
  stream_action('remove', items: items, packets: packets, scope: scope)
end

#stream_action(action, items:, packets:, scope:, extra: {}) ⇒ Object

Build and write a StreamingChannel action. extra carries action specific keys (the times for 'add') and is merged first to preserve wire ordering.



533
534
535
536
537
538
539
540
541
542
 
# File 'lib/openc3/script/web_socket_api.rb', line 533

def stream_action(action, items:, packets:, scope:, extra: {})
  data_hash = {}
  data_hash['action'] = action
  data_hash.merge!(extra)
  data_hash['items'] = items if items
  data_hash['packets'] = packets if packets
  data_hash['scope'] = scope || @scope
  data_hash['token'] = @authentication.token(include_bearer: false)
  write_action(data_hash)
end

#to_nsec(value) ⇒ Object

Accept either a Time or an already converted 64-bit nanosecond value



527
528
529
 
# File 'lib/openc3/script/web_socket_api.rb', line 527

def to_nsec(value)
  return Time === value ? value.to_nsec_from_epoch : value
end