Method: ContentServer::FileCopyServer#run

Defined in:

lib/content_server/queue_copy.rb

#run ⇒ Object

# File 'lib/content_server/queue_copy.rb', line 165

def run()
  threads = []
  threads << @backup_tcp.tcp_thread if @backup_tcp != nil
  threads << Thread.new do
    while true do
      Log.debug1 'Waiting on copy files events.'
      (message_type, message_content) = @copy_input_queue.pop
      $process_vars.set('Copy File Queue Size', @copy_input_queue.size)
      Log.debug1("Content copy message:Type:%s  content:%s", message_type, message_content)

      if message_type == :SEND_ACK_MESSAGE
        Log.debug1("Sending ack for: %s", message_content)
        @backup_tcp.send_obj([:ACK_MESSAGE, [message_content, Time.now.to_i]])
      elsif message_type == :COPY_MESSAGE
        message_content.each_instance { |checksum, size, content_mod_time, instance_mod_time, server, path|
          @file_copy_manager.add_content(checksum, path)
        }
      elsif message_type == :ACK_MESSAGE
        # Received ack from backup, copy file if all is good.
        # The timestamp is of local content server! not backup server!
        timestamp, ack, checksum = message_content
        Log.debug1("Ack (#{ack}) received for content: #{checksum}, timestamp: #{timestamp} " \
                   "now: #{Time.now.to_i}") if Params['log_debug_level'] >= 1  # adding to avoid Time.now

        # Copy file if ack (does not exists on backup and not too much time passed)
        if ack
          if (Time.now.to_i - timestamp < Params['ack_timeout'])
            @file_copy_manager.receive_ack(checksum)
          else
            Log.debug1("Ack timed out span: #{Time.now.to_i - timestamp} > " \
              "timeout: #{Params['ack_timeout']}") if Params['log_debug_level'] >= 1  # adding to avoid Time.now
            # remove only content under copy
            @file_copy_manager.remove_content(checksum)
          end
        else
          Log.debug1('Ack is false');
          # remove content under copy and content in waiting list
          @file_copy_manager.remove_content(checksum)
        end
      elsif message_type == :COPY_CHUNK_FROM_REMOTE
        checksum = message_content
        @file_streamer.copy_another_chuck(checksum)
      elsif message_type == :COPY_CHUNK
        # We open the message here for printing info and deleting copy_prepare!
        file_checksum, offset, file_size, content, content_checksum = message_content
        Log.debug1("Send chunk for file %s, offset: %s " \
                     "filesize: %s, checksum: %s",
                   file_checksum, offset, file_size, content_checksum)
        # Blocking send.
        @backup_tcp.send_obj([:COPY_CHUNK, message_content])
        if content.nil? and content_checksum.nil?
          # Sending enf of file and removing file from list
          @file_copy_manager.remove_content(file_checksum)
        end
      elsif message_type == :ABORT_COPY
        Log.debug1("Aborting file copy: %s", message_content)
        @file_streamer.abort_streaming(message_content)
        # remove only content under copy
        @file_copy_manager.remove_content(message_content)
      elsif message_type == :RESET_RESUME_COPY
        (file_checksum, new_offset) = message_content
        Log.debug1("Resetting/Resuming file (%s) copy to %s",
                   file_checksum, new_offset)
        @file_streamer.reset_streaming(file_checksum, new_offset)
      else
        Log.error("Copy event not supported: #{message_type}")
      end # handle messages here
    end
    Log.error("Should not reach here, loop should continue.")
  end
end