Class: Aspera::Fasp::AgentNode

Inherits:
AgentBase show all
Defined in:
lib/aspera/fasp/agent_node.rb

Overview

this singleton class is used by the CLI to provide a common interface to start a transfer before using it, the use must set the node_api member.

Constant Summary

Constants inherited from AgentBase

Aspera::Fasp::AgentBase::LISTENER_SESSION_ID_B, Aspera::Fasp::AgentBase::LISTENER_SESSION_ID_S

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from AgentBase

#add_listener, validate_status_list

Constructor Details

#initialize(options) ⇒ AgentNode 



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/aspera/fasp/agent_node.rb', line 17

def initialize(options)
  raise 'node specification must be Hash' unless options.is_a?(Hash)
  i[url username password].each { |k| raise "missing parameter [#{k}] in node specification: #{options}" unless options.key?(k) }
  super()
  # root id is required for access key
  @root_id = options[:root_id]
  rest_params = { base_url: options[:url]}
  if /^Bearer /.match?(options[:password])
    rest_params[:headers] = {
      Aspera::Node::HEADER_X_ASPERA_ACCESS_KEY => options[:username],
      'Authorization'                          => options[:password]
    }
    raise 'root_id is required for access key' if @root_id.nil?
  else
    rest_params[:auth] = {
      type:     :basic,
      username: options[:username],
      password: options[:password]
    }
  end
  @node_api = Rest.new(rest_params)
  # TODO: currently only supports one transfer. This is bad shortcut. but ok for CLI.
  @transfer_id = nil
end

Instance Attribute Details

#node_api ⇒ Object

use this to read the node_api end point.



48
49
50
 
# File 'lib/aspera/fasp/agent_node.rb', line 48

def node_api
  @node_api
end

#options=(value) ⇒ Object (writeonly)

option include: root_id if the node is an access key



15
16
17
 
# File 'lib/aspera/fasp/agent_node.rb', line 15

def options=(value)
  @options = value
end

Instance Method Details

#node_api_ ⇒ Object

used internally to ensure node api is set before using.

Raises:

  • (StandardError) 


43
44
45
46
 
# File 'lib/aspera/fasp/agent_node.rb', line 43

def node_api_
  raise StandardError, 'Before using this object, set the node_api attribute to a Aspera::Rest object' if @node_api.nil?
  return @node_api
end

#start_transfer(transfer_spec, token_regenerator: nil) ⇒ Object

generic method



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/aspera/fasp/agent_node.rb', line 59

def start_transfer(transfer_spec, token_regenerator: nil)
  # add root id if access key
  if !@root_id.nil?
    case transfer_spec['direction']
    when Fasp::TransferSpec::DIRECTION_SEND then transfer_spec['source_root_id'] = @root_id
    when Fasp::TransferSpec::DIRECTION_RECEIVE then transfer_spec['destination_root_id'] = @root_id
    else raise "unexpected direction in ts: #{transfer_spec['direction']}"
    end
  end
  # manage special additional parameter
  if transfer_spec.key?('EX_ssh_key_paths') && transfer_spec['EX_ssh_key_paths'].is_a?(Array) && !transfer_spec['EX_ssh_key_paths'].empty?
    # not standard, so place standard field
    if transfer_spec.key?('ssh_private_key')
      Log.log.warn('Both ssh_private_key and EX_ssh_key_paths are present, using ssh_private_key')
    else
      Log.log.warn('EX_ssh_key_paths has multiple keys, using first one only') unless transfer_spec['EX_ssh_key_paths'].length.eql?(1)
      transfer_spec['ssh_private_key'] = File.read(transfer_spec['EX_ssh_key_paths'].first)
      transfer_spec.delete('EX_ssh_key_paths')
    end
  end
  # add mandatory retry parameter for node api
  ts_tags = transfer_spec['tags']
  if ts_tags.is_a?(Hash) && ts_tags[Fasp::TransferSpec::TAG_RESERVED].is_a?(Hash)
    ts_tags[Fasp::TransferSpec::TAG_RESERVED]['xfer_retry'] ||= 150
  end
  # Optimization in case of sending to the same node
  # TODO: probably remove this, as /etc/hosts shall be used for that
  if !transfer_spec['wss_enabled'] && transfer_spec['remote_host'].eql?(URI.parse(node_api_.params[:base_url]).host)
    transfer_spec['remote_host'] = '127.0.0.1'
  end
  resp = node_api_.create('ops/transfers', transfer_spec)[:data]
  @transfer_id = resp['id']
  Log.log.debug{"tr_id=#{@transfer_id}"}
  return @transfer_id
end

#wait_for_transfers_completion ⇒ Object

generic method



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
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/aspera/fasp/agent_node.rb', line 96

def wait_for_transfers_completion
  started = false
   = nil
  # lets emulate management events to display progress bar
  loop do
    # status is empty sometimes with status 200...
    transfer_data = node_api_.read("ops/transfers/#{@transfer_id}")[:data] || {'status' => 'unknown'} rescue {'status' => 'waiting(read error)'}
    case transfer_data['status']
    when 'completed'
      notify_end(@transfer_id)
      break
    when 'waiting', 'partially_completed', 'unknown', 'waiting(read error)'
      if .nil?
         = TTY::Spinner.new('[:spinner] :title', format: :classic)
        .start
      end
      .update(title: transfer_data['status'])
      .spin
    when 'running'
      if !started && transfer_data['precalc'].is_a?(Hash) &&
          transfer_data['precalc']['status'].eql?('ready')
        notify_begin(@transfer_id, transfer_data['precalc']['bytes_expected'])
        started = true
      else
        notify_progress(@transfer_id, transfer_data['bytes_transferred'])
      end
    when 'failed'
      # Bug in HSTS ? transfer is marked failed, but there is no reason
      if transfer_data['error_code'].eql?(0) && transfer_data['error_desc'].empty?
        notify_end(@transfer_id)
        break
      end
      raise Fasp::Error, "status: #{transfer_data['status']}. code: #{transfer_data['error_code']}. description: #{transfer_data['error_desc']}"
    else
      Log.log.warn{"transfer_data -> #{transfer_data}"}
      raise Fasp::Error, "status: #{transfer_data['status']}. code: #{transfer_data['error_code']}. description: #{transfer_data['error_desc']}"
    end
    sleep(1)
  end
  # TODO: get status of sessions
  return []
end