Class: Cosmos::InterfaceMicroservice

Inherits:
Microservice show all
Defined in:
lib/cosmos/microservices/interface_microservice.rb

Direct Known Subclasses

RouterMicroservice

Constant Summary collapse

UNKNOWN_BYTES_TO_PRINT = 
16

Instance Attribute Summary

Attributes inherited from Microservice

#count, #custom, #error, #microservice_status_thread, #name, #scope, #state

Instance Method Summary collapse

Methods inherited from Microservice

#as_json, run

Constructor Details

#initialize(name) ⇒ InterfaceMicroservice

Returns a new instance of InterfaceMicroservice.



246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 246

def initialize(name)
  super(name)
  @interface_or_router = self.class.name.to_s.split("Microservice")[0].upcase.split("::")[-1]
  @scope = name.split("__")[0]
  interface_name = name.split("__")[2]
  if @interface_or_router == 'INTERFACE'
    @interface = InterfaceModel.get_model(name: interface_name, scope: @scope).build
  else
    @interface = RouterModel.get_model(name: interface_name, scope: @scope).build
  end
  @interface.name = interface_name
  # Map the interface to the interface's targets
  @interface.target_names do |target_name|
    target = System.targets[target_name]
    target.interface = @interface
  end
  if @interface.connect_on_startup
    @interface.state = 'ATTEMPTING'
  else
    @interface.state = 'DISCONNECTED'
  end
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json, scope: @scope)
  end

  @interface_thread_sleeper = Sleeper.new
  @cancel_thread = false
  @connection_failed_messages = []
  @connection_lost_messages = []
  @mutex = Mutex.new
  if @interface_or_router == 'INTERFACE'
    @handler_thread = InterfaceCmdHandlerThread.new(@interface, self, scope: @scope)
  else
    @handler_thread = RouterTlmHandlerThread.new(@interface, self, scope: @scope)
  end
  @handler_thread.start
end

Instance Method Details

#attempting ⇒ Object

External method to be called by the InterfaceCmdHandlerThread to connect Thus we just set the state and allow the run method to handle the action



288
289
290
291
292
293
294
295
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 288

def attempting
  @interface.state = 'ATTEMPTING'
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json, scope: @scope)
  end
end

#connect ⇒ Object 



472
473
474
475
476
477
478
479
480
481
482
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 472

def connect
  Logger.info "#{@interface.name}: Connecting ..."
  @interface.connect
  @interface.state = 'CONNECTED'
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json, scope: @scope)
  end
  Logger.info "#{@interface.name}: Connection Success"
end

#disconnect(allow_reconnect = true) ⇒ Object 



484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 484

def disconnect(allow_reconnect = true)
  return if @interface.state == 'DISCONNECTED' && !@interface.connected?

  # Synchronize the calls to @interface.disconnect since it takes an unknown
  # amount of time. If two calls to disconnect stack up, the if statement
  # should avoid multiple calls to disconnect.
  @mutex.synchronize do
    begin
      @interface.disconnect if @interface.connected?
    rescue => e
      Logger.error "Disconnect: #{@interface.name}: #{e.formatted}"
    end
  end

  # If the interface is set to auto_reconnect then delay so the thread
  # can come back around and allow the interface a chance to reconnect.
  if allow_reconnect and @interface.auto_reconnect and @interface.state != 'DISCONNECTED'
    attempting()
    if !@cancel_thread
      # Logger.debug "reconnect delay: #{@interface.reconnect_delay}"
      @interface_thread_sleeper.sleep(@interface.reconnect_delay)
    end
  else
    @interface.state = 'DISCONNECTED'
    if @interface_or_router == 'INTERFACE'
      InterfaceStatusModel.set(@interface.as_json, scope: @scope)
    else
      RouterStatusModel.set(@interface.as_json, scope: @scope)
    end
  end
end

#graceful_kill ⇒ Object 



541
542
543
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 541

def graceful_kill
  # Just to avoid warning
end

#handle_connection_failed(connect_error) ⇒ Object 



426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 426

def handle_connection_failed(connect_error)
  @error = connect_error
  Logger.error "#{@interface.name}: Connection Failed: #{connect_error.formatted(false, false)}"
  case connect_error
  when Interrupt
    Logger.info "#{@interface.name}: Closing from signal"
    @cancel_thread = true
  when Errno::ECONNREFUSED, Errno::ECONNRESET, Errno::ETIMEDOUT, Errno::ENOTSOCK, Errno::EHOSTUNREACH, IOError
    # Do not write an exception file for these extremely common cases
  else
    if RuntimeError === connect_error and (connect_error.message =~ /canceled/ or connect_error.message =~ /timeout/)
      # Do not write an exception file for these extremely common cases
    else
      Logger.error "#{@interface.name}: #{connect_error.formatted}"
      unless @connection_failed_messages.include?(connect_error.message)
        Cosmos.write_exception_file(connect_error)
        @connection_failed_messages << connect_error.message
      end
    end
  end
  disconnect() # Ensure we do a clean disconnect
end

#handle_connection_lost(err = nil, reconnect: true) ⇒ Object 



449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 449

def handle_connection_lost(err = nil, reconnect: true)
  if err
    @error = err
    Logger.info "#{@interface.name}: Connection Lost: #{err.formatted(false, false)}"
    case err
    when Interrupt
      Logger.info "#{@interface.name}: Closing from signal"
      @cancel_thread = true
    when Errno::ECONNABORTED, Errno::ECONNRESET, Errno::ETIMEDOUT, Errno::EBADF, Errno::ENOTSOCK, IOError
      # Do not write an exception file for these extremely common cases
    else
      Logger.error "#{@interface.name}: #{err.formatted}"
      unless @connection_lost_messages.include?(err.message)
        Cosmos.write_exception_file(err)
        @connection_lost_messages << err.message
      end
    end
  else
    Logger.info "#{@interface.name}: Connection Lost"
  end
  disconnect(reconnect) # Ensure we do a clean disconnect
end

#handle_packet(packet) ⇒ Object 



361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 361

def handle_packet(packet)
  InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  packet.received_time = Time.now.sys unless packet.received_time

  if packet.stored
    # Stored telemetry does not update the current value table
    identified_packet = System.telemetry.identify_and_define_packet(packet, @target_names)
  else
    # Identify and update packet
    if packet.identified?
      begin
        # Preidentifed packet - place it into the current value table
        identified_packet = System.telemetry.update!(packet.target_name,
                                                     packet.packet_name,
                                                     packet.buffer)
      rescue RuntimeError
        # Packet identified but we don't know about it
        # Clear packet_name and target_name and try to identify
        Logger.warn "#{@interface.name}: Received unknown identified telemetry: #{packet.target_name} #{packet.packet_name}"
        packet.target_name = nil
        packet.packet_name = nil
        identified_packet = System.telemetry.identify!(packet.buffer,
                                                       @target_names)
      end
    else
      # Packet needs to be identified
      identified_packet = System.telemetry.identify!(packet.buffer,
                                                     @target_names)
    end
  end

  if identified_packet
    identified_packet.received_time = packet.received_time
    identified_packet.stored = packet.stored
    identified_packet.extra = packet.extra
    packet = identified_packet
  else
    unknown_packet = System.telemetry.update!('UNKNOWN', 'UNKNOWN', packet.buffer)
    unknown_packet.received_time = packet.received_time
    unknown_packet.stored = packet.stored
    unknown_packet.extra = packet.extra
    packet = unknown_packet
    json_hash = CvtModel.build_json_from_packet(packet)
    CvtModel.set(json_hash, target_name: packet.target_name, packet_name: packet.packet_name, scope: scope)
    num_bytes_to_print = [UNKNOWN_BYTES_TO_PRINT, packet.length].min
    data = packet.buffer(false)[0..(num_bytes_to_print - 1)]
    prefix = data.each_byte.map { | byte | sprintf("%02X", byte) }.join()
    Logger.warn "#{@interface.name} #{packet.target_name} packet length: #{packet.length} starting with: #{prefix}"
  end

  # Write to stream
  packet.received_count += 1
  TelemetryTopic.write_packet(packet, scope: @scope)
end

#inject_tlm(hash) ⇒ Object 



416
417
418
419
420
421
422
423
424
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 416

def inject_tlm(hash)
  packet = System.telemetry.packet(hash['target_name'], hash['packet_name']).clone
  if hash['item_hash']
    JSON.parse(hash['item_hash']).each do |item, value|
      packet.write(item.to_s, value, hash['type'].to_sym)
    end
  end
  handle_packet(packet)
end

#run ⇒ Object 



297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 297

def run
  begin
    if @interface.read_allowed?
      Logger.info "#{@interface.name}: Starting packet reading"
    else
      Logger.info "#{@interface.name}: Starting connection maintenance"
    end
    while true
      break if @cancel_thread

      case @interface.state
      when 'DISCONNECTED'
        begin
          # Just wait to see if we should connect later
          @interface_thread_sleeper.sleep(1)
        rescue Exception => err
          break if @cancel_thread
        end
      when 'ATTEMPTING'
        begin
          @mutex.synchronize do
            # We need to make sure connect is not called after stop() has been called
            connect() unless @cancel_thread
          end
        rescue Exception => connect_error
          handle_connection_failed(connect_error)
          break if @cancel_thread
        end
      when 'CONNECTED'
        if @interface.read_allowed?
          begin
            packet = @interface.read
            if packet
              handle_packet(packet)
              @count += 1
            else
              Logger.info "#{@interface.name}: Internal disconnect requested (returned nil)"
              handle_connection_lost()
              break if @cancel_thread
            end
          rescue Exception => err
            handle_connection_lost(err)
            break if @cancel_thread
          end
        else
          @interface_thread_sleeper.sleep(1)
          handle_connection_lost() if !@interface.connected?
        end
      end
    end
  rescue Exception => error
    Logger.error "#{@interface.name}: Packet reading thread died: #{error.formatted}"
    Cosmos.handle_fatal_exception(error)
    # Try to do clean disconnect because we're going down
    disconnect(false)
  end
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json, scope: @scope)
  end
  Logger.info "#{@interface.name}: Stopped packet reading"
end

#shutdown(sig = nil) ⇒ Object 



535
536
537
538
539
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 535

def shutdown(sig = nil)
  Logger.info "#{@interface.name}: shutdown requested"
  stop()
  super()
end

#stop ⇒ Object

Disconnect from the interface and stop the thread



517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
 
# File 'lib/cosmos/microservices/interface_microservice.rb', line 517

def stop
  Logger.info "#{@interface.name}: stop requested"
  @mutex.synchronize do
    # Need to make sure that @cancel_thread is set and the interface disconnected within
    # mutex to ensure that connect() is not called when we want to stop()
    @cancel_thread = true
    @handler_thread.stop
    @interface_thread_sleeper.cancel
    @interface.disconnect
    if @interface_or_router == 'INTERFACE'
      valid_interface = InterfaceStatusModel.get_model(name: @interface.name, scope: @scope)
    else
      valid_interface = RouterStatusModel.get_model(name: @interface.name, scope: @scope)
    end
    valid_interface.destroy if valid_interface
  end
end