Class: Conversant::V3::Services::LMS::Job

Inherits:
Object
  • Object
show all
Defined in:
lib/conversant/v3/services/lms/job.rb,
sig/conversant/v3/services/lms.rbs

Overview

Job management for streaming

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(parent) ⇒ Job

Initialize job service

Parameters:

  • parent (LMS) —

    the parent LMS service instance

Since:

  • 1.0.0



35
36
37
 
# File 'lib/conversant/v3/services/lms/job.rb', line 35

def initialize(parent)
  @parent = parent
end

Instance Attribute Details

#parent ⇒ LMS (readonly)

Returns the parent LMS service instance.

Returns:

  • (LMS) —

    the parent LMS service instance

Since:

  • 1.0.0



30
31
32
 
# File 'lib/conversant/v3/services/lms/job.rb', line 30

def parent
  @parent
end

Instance Method Details

#build_manifest_url(play_domain, item, output_type) ⇒ String?

Build manifest URL for a job

Parameters:

  • play_domain (String) —

    playback domain

  • item (Hash) —

    job item hash

  • output_type (String) —

    output stream type ('hls', 'dash')

Returns:

  • (String, nil) —

    manifest URL or nil if play_domain is missing

Since:

  • 1.0.0



188
189
190
191
192
193
194
195
196
 
# File 'lib/conversant/v3/services/lms/job.rb', line 188

def build_manifest_url(play_domain, item, output_type)
  return nil unless play_domain

  app = item['app'] || item[:app]
  manifest_name = item['manifest_name'] || item[:manifest_name]
  ext = stream_extension(output_type)

  "https://#{play_domain}/#{app}/#{manifest_name}#{ext}"
end

#extract_encoder(item) ⇒ String

Extract encoder IP from stream metadata

Parameters:

  • item (Hash) —

    job item hash

Returns:

  • (String) —

    encoder IP address or '-' if not found

Since:

  • 1.0.0



165
166
167
168
 
# File 'lib/conversant/v3/services/lms/job.rb', line 165

def extract_encoder(item)
  metadata = item['stream_metadata'] || item[:stream_metadata]
  metadata && (metadata['source_ip'] || metadata[:source_ip]) || '-'
end

#parse_status(status) ⇒ String

Parse numeric status code to string

Parameters:

  • status (Integer) —

    numeric status code

Returns:

  • (String) —

    status string or 'unknown_N' if unknown

Since:

  • 1.0.0



134
135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/conversant/v3/services/lms/job.rb', line 134

def parse_status(status)
  case status
  when 1 then 'ready'
  when 2 then 'streaming'
  when 3 then 'stopped'
  when 4 then 'error'
  when 5 then 'starting'
  when 6 then 'stopping'
  when 7 then 'restarting'
  when 8 then 'pending'
  when 12 then 'interrupted'
  else "unknown_#{status}"
  end
end

#parse_timestamp(timestamp) ⇒ DateTime?

Parse Unix timestamp (milliseconds) to DateTime

Parameters:

  • timestamp (Integer) —

    Unix timestamp in milliseconds

Returns:

  • (DateTime, nil) —

    parsed DateTime or nil if invalid

Since:

  • 1.0.0



153
154
155
156
157
158
159
 
# File 'lib/conversant/v3/services/lms/job.rb', line 153

def parse_timestamp(timestamp)
  return nil unless timestamp

  Time.at(timestamp.to_i / 1000).to_datetime
rescue StandardError
  nil
end

#stream_extension(type) ⇒ String

Get file extension for stream type

Parameters:

  • type (String) —

    stream type ('hls', 'dash')

Returns:

  • (String) —

    file extension with dot

Since:

  • 1.0.0



174
175
176
177
178
179
180
 
# File 'lib/conversant/v3/services/lms/job.rb', line 174

def stream_extension(type)
  case type
  when 'hls' then '.m3u8'
  when 'dash' then '.mpd'
  else ''
  end
end

#where(**params) ⇒ Array<Hash>?

Query streaming jobs with filters

Retrieves a list of streaming jobs matching the specified filters. Returns detailed information about each job including status, profiles, and manifest URLs.

rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity

Examples:

Get streaming jobs

jobs = lms.job.where(status: 'streaming', limit: 50)
jobs.each do |job|
  puts "#{job[:name]} (#{job[:status]}) - #{job[:manifest]}"
end

Get all jobs

all_jobs = lms.job.where

Parameters:

  • params (Hash) —

    query parameters

Options Hash (**params):

  • :status (String) —

    job status filter ('streaming', 'ready', 'interrupted')

  • :limit (Integer) —

    maximum number of results

  • :offset (Integer) —

    pagination offset

Returns:

  • (Array<Hash>, nil) —

    array of job hashes with details, or nil on error Each hash contains:

    • :entity [Integer] OSP entity ID
    • :id [Integer] job ID
    • :app [String] application name
    • :name [String] stream name
    • :status [String] job status ('ready', 'streaming', 'interrupted')
    • :created_at [DateTime] job creation timestamp
    • :started_at [DateTime] job start timestamp
    • :ended_at [DateTime] job end timestamp
    • :client [String] client IP address
    • :encoder [String] encoder IP address
    • :server [String] execution server
    • :manifest [String] manifest URL
    • :profiles_count [Integer] number of stream profiles
    • :transcoding [Boolean] transcoding enabled
    • :gpu [Boolean] GPU acceleration enabled
    • :ha [Boolean] high availability enabled
    • :dvr [Boolean] DVR enabled
    • :dai [Boolean] dynamic ad insertion enabled
    • :encrypted [Boolean] stream encryption enabled
    • :ingest_url [String] RTMP ingest URL
    • :play_domain [String] playback domain

Since:

  • 1.0.0



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/conversant/v3/services/lms/job.rb', line 85

def where(**params)
  response = JSON.parse(@parent.send(:call, 'GET', "/v4/streams?#{params.to_query}"))
  return nil unless response

  items = [response['list'] || response[:list] || []].flatten.map(&:with_indifferent_access)

  items.map do |item|
    status = item[:status]&.to_i
    inbound = item[:stream_rtmp_inbound]
    play_domain = item[:play_domain] || item['play_domain']
    outbounds = item[:stream_rtmp_outbounds] || item['stream_rtmp_outbounds'] || []
    output_type = outbounds.first && (outbounds.first['type'] || outbounds.first[:type])

    {
      entity: item['osp_id'] || item[:osp_id],
      id: item['id'] || item[:id],
      app: item['app'] || item[:app],
      name: item['stream'] || item[:stream],
      status: parse_status(status),
      created_at: parse_timestamp(item['created'] || item[:created]),
      started_at: parse_timestamp(item['started'] || item[:started]),
      ended_at: parse_timestamp(item['ended'] || item[:ended]),
      client: inbound && (inbound['client_ip'] || inbound[:client_ip]),
      ingest_url: inbound&.dig(:push_url) || inbound&.dig('push_url'),
      encoder: extract_encoder(item),
      server: item['exec_ta'] || item[:exec_ta],
      manifest: build_manifest_url(play_domain, item, output_type),
      profiles_count: outbounds.size,
      transcoding: item['transcoding'] || item[:transcoding],
      gpu: (item['gpu'] || item[:gpu]) == 'gpu',
      ha: item['ha'] || item[:ha],
      dvr: item['dvr'] || item[:dvr],
      dai: item['dai'] || item[:dai],
      encrypted: item[:stream_encrypts]&.any? || item['stream_encrypts']&.any?,
      play_domain: play_domain
    }
  end
rescue StandardError => e
  @parent.send(:logger).error "LMS::Job.where error: #{e.message}"
  nil
end