Class: MusicStory::Repository::BatchSFTP

Inherits:
Object
  • Object
show all
Defined in:
lib/music_story/repository/batch_sftp.rb

Overview

Gives access to whole batches of music story data that are kept somewhere on an sftp site somwhere that music story can constantly deliver new data. Once downloaded, this data can be accessed using the xml repository, or imported in to a local database

Defined Under Namespace

Classes: SessionWrapper

Constant Summary collapse

DELIVERY_COMPLETE =

the presence of this file inside a batch directory tells us that the sender has finished sending it

'delivery.complete'
DIR_PROCESSING =

some sub dirs we use to manage the flow of data

'processing'
DIR_PROCESSED = 
'processed'
RENAME_NATIVE =

memoised flag from sftp rename methods

Net::SFTP::Constants::RenameFlags::NATIVE

Instance Method Summary collapse

Constructor Details

#initialize(host, username, options = {}) ⇒ BatchSFTP

Returns a new instance of BatchSFTP.



23
24
25
26
27
28
29
30
 
# File 'lib/music_story/repository/batch_sftp.rb', line 23

def initialize(host, username, options={})
  @host = host
  @username = username
  @options = options
  @basedir = options[:basedir] || '/'
  @batch_pattern = options[:batch_pattern] || 'music-story-data-*'
  @logger = options[:logger] || Logger.new('/dev/null')
end

Instance Method Details

#connect(&block) ⇒ Object

start talking to the remote server, yielding the session to the block, which is closed after the block finishes executing. The block is yielded a wrapper object that lets you use the access methods in the repository, minus the first argument, for instance:

repo.connect do |session|
  batch = session.new_batches.first
  session.download(batch, '/tmp/dir')
end



40
41
42
43
44
45
46
47
 
# File 'lib/music_story/repository/batch_sftp.rb', line 40

def connect(&block)
  return_result = nil
  # the sftp.start method does not seem to return the last thing you execute
  start_sftp_session do |sftp_session|
    return_result = yield SessionWrapper.new(self, sftp_session)
  end
  return_result
end

#download(w, batch, local_dir) ⇒ Object

download a batch. Should work for a batch in any state



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/music_story/repository/batch_sftp.rb', line 106

def download(w, batch, local_dir)
  @logger.info("Downloading #{batch.path} to #{local_dir}...")
  w.sftp.download!(batch.path, local_dir, :recursive => true) do |event, downloader, *args|
    case event
    when :open then
      # args[0] : file metadata
      @logger.debug "Starting download: #{args[0].remote} -> #{args[0].local} (#{args[0].size}) bytes"
    when :close then
      # args[0] : file metadata
      @logger.debug "Finished download: #{args[0].remote}"
    when :finish then
      @logger.debug "Download complete"
    end
  end
end

#mark_processed(w, batch, path_to_logfile = nil) ⇒ Object

move a batch in to the processed state, moving its location on the remote fs



140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/music_story/repository/batch_sftp.rb', line 140

def mark_processed(w, batch, path_to_logfile=nil)
  batch_basename = File.basename(batch.path)
  new_name = join(@basedir, DIR_PROCESSED, batch_basename)
  @logger.info("Marking #{batch} as processed")
  @logger.debug("  moving #{batch.path} to #{new_name}")
  w.sftp.rename(batch.path, new_name, RENAME_NATIVE)
  batch.path = new_name
  batch.state = :processed

  if path_to_logfile
    remote_logfile_path = join(@basedir, batch_basename + '.log')
    uploader = w.sftp.upload!(path_to_logfile, remote_logfile_path)
  end
end

#mark_processing(w, batch) ⇒ Object

move a batch in to the processing state, moving its location on the remote fs



129
130
131
132
133
134
135
136
 
# File 'lib/music_story/repository/batch_sftp.rb', line 129

def mark_processing(w, batch)
  new_name = join(@basedir, DIR_PROCESSING, File.basename(batch.path))
  @logger.info("Marking #{batch} as processing")
  @logger.debug("  moving #{batch.path} to #{new_name}")
  w.sftp.rename(batch.path, new_name, RENAME_NATIVE)
  batch.path = new_name
  batch.state = :processing
end

#new_available?(w) ⇒ Boolean

return true if there are any batches available

Returns:

  • (Boolean) 


123
124
125
 
# File 'lib/music_story/repository/batch_sftp.rb', line 123

def new_available?(w)
  new_batches(w).any?
end

#new_batches(w) ⇒ Object

return a list of all the batches on the sftp site that are ready to be downloaded or we can start processing them



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/music_story/repository/batch_sftp.rb', line 85

def new_batches(w)
  @logger.debug("Looking for new batches in remote dir '#@basedir' with pattern #@batch_pattern")
  complete_dirs = w.sftp.dir[@basedir, @batch_pattern].select do |entry|
    next if /\.log$/.match(entry.name) # skip log files (MSP#1915)

    w.sftp.dir[join(@basedir, entry.name), DELIVERY_COMPLETE].any?.tap do |f|
      if f
        @logger.debug("  Found new batch: #{entry.name}")
      else
        @logger.debug("  Incomplete batch: #{entry.name}")
      end
    end
  end

  complete_dirs.map do |entry|
    Model::Batch.new(:path => join(@basedir, entry.name),
      :state => :new)
  end
end

#processed_batches(w) ⇒ Object

return a list of batches on the sftp that are in the processed state, i.e live in the ‘processed` directory



67
68
69
70
71
72
 
# File 'lib/music_story/repository/batch_sftp.rb', line 67

def processed_batches(w)
  dir = join(@basedir, DIR_PROCESSED)
  w.sftp.dir[dir, '*'].map do |entry|
    Model::Batch.new(:path => join(dir, entry.name), :state => :processed)
  end
end

#processing_batches(w) ⇒ Object

return a list of batches on the sftp site that are in the processing state, i.e live in the ‘processing` directory



76
77
78
79
80
81
 
# File 'lib/music_story/repository/batch_sftp.rb', line 76

def processing_batches(w)
  dir = join(@basedir, DIR_PROCESSING)
  w.sftp.dir[dir, '*'].map do |entry|
    Model::Batch.new(:path => join(dir, entry.name), :state => :processing)
  end
end

#start_sftp_session(&block) ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/music_story/repository/batch_sftp.rb', line 49

def start_sftp_session(&block)
  cnx_options = (@options[:net_sftp_options] || {}).
    merge(:password => @options[:password])

  @logger.info("Starting sftp session to '#{@host}'")
  Net::SFTP.start(@host, @username, cnx_options) do |sftp_session|
    block.call(sftp_session)
  end.tap do
    @logger.info("Finished sftp session to '#{@host}'")
  end
end