Class: Aspera::Cli::AsyncTransferStore

Inherits:
Object
  • Object
show all
Defined in:
lib/aspera/cli/async_transfer_store.rb

Overview

Persist the state of asynchronous transfers on the local file system.

Each entry is a JSON document identified by a job_id (UUID generated by ascli). The underlying storage uses PersistencyFolder with the category prefix CATEGORY so all async-transfer files are grouped together and can be garbage-collected independently.

Schema of a stored entry:

job_id            [String]  UUID generated by ascli (= the store key)
agent_type        [String]  'desktop' | 'node' | 'connect' | 'transferd' | 'direct'
transfer_id       [String]  Opaque ID returned by the agent's start_transfer
agent_params      [Hash]    Agent-specific connection parameters for re-querying
status            [String]  'running' | 'completed' | 'failed' | 'cancelled'
bytes_transferred [Integer] Last known bytes transferred (0 if unknown)
started_at        [String]  ISO-8601 timestamp
ended_at          [String, nil] ISO-8601 timestamp once finished
error             [String, nil] Error message when status == 'failed'

Class Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(persistency) ⇒ AsyncTransferStore

Returns a new instance of AsyncTransferStore.

Parameters:



41
42
43
 
# File 'lib/aspera/cli/async_transfer_store.rb', line 41

def initialize(persistency)
  @persistency = persistency
end

Class Attribute Details

.agent_refs ⇒ Object (readonly)

Returns the value of attribute agent_refs.



37
38
39
 
# File 'lib/aspera/cli/async_transfer_store.rb', line 37

def agent_refs
  @agent_refs
end

Instance Method Details

#agent_ref(job_id) ⇒ Object?

Retrieve a previously registered in-process agent reference.

Parameters:

Returns:

  • (Object, nil) —

    the live agent instance, or nil if not registered / process restarted



57
58
59
 
# File 'lib/aspera/cli/async_transfer_store.rb', line 57

def agent_ref(job_id)
  self.class.agent_refs[job_id]
end

#delete(job_id) ⇒ Object

Delete one entry.

Parameters:



95
96
97
98
99
 
# File 'lib/aspera/cli/async_transfer_store.rb', line 95

def delete(job_id)
  Aspera.assert_type(job_id, String) { 'job_id' }
  @persistency.delete(store_key(job_id))
  nil
end

#list ⇒ Array<Hash>

List all async transfer entries.

Returns:

  • (Array<Hash>) —

    each entry includes the job_id field



85
86
87
88
89
90
91
 
# File 'lib/aspera/cli/async_transfer_store.rb', line 85

def list
  @persistency.current_items(CATEGORY).map do |key, raw|
    data = JSON.parse(raw)
    data['job_id'] ||= key.sub(CATEGORY, '')
    data
  end
end

#read(job_id) ⇒ Hash?

Read one entry.

Parameters:

Returns:

  • (Hash, nil) —

    the stored data, or nil if not found



76
77
78
79
80
81
 
# File 'lib/aspera/cli/async_transfer_store.rb', line 76

def read(job_id)
  Aspera.assert_type(job_id, String) { 'job_id' }
  raw = @persistency.get(store_key(job_id))
  return if raw.nil?
  JSON.parse(raw)
end

#register_agent_ref(job_id, agent_ref) ⇒ Object

Register an in-process agent reference for a job. Called by TransferAgent immediately after start_transfer (async mode). The reference is only available while the same Ruby process is alive.

Parameters:

  • job_id (String) —

    the ascli-generated UUID

  • agent_ref (Object) —

    live agent instance (responds to sessions_by_job)



50
51
52
 
# File 'lib/aspera/cli/async_transfer_store.rb', line 50

def register_agent_ref(job_id, agent_ref)
  self.class.agent_refs[job_id] = agent_ref
end

#write(job_id, data) ⇒ Object

Persist (create or update) an async transfer entry. Keys whose name starts with '_' are in-process references (e.g. '_agent_ref') and are intentionally excluded from the JSON serialization — at every nesting level.

Parameters:

  • job_id (String) —

    the ascli-generated UUID

  • data (Hash) —

    fields to store (will be JSON-serialised)



66
67
68
69
70
71
 
# File 'lib/aspera/cli/async_transfer_store.rb', line 66

def write(job_id, data)
  Aspera.assert_type(job_id, String) { 'job_id' }
  Aspera.assert_type(data, Hash) { 'data' }
  @persistency.put(store_key(job_id), JSON.generate(strip_internal_keys(data)))
  nil
end