Class: ContentServer::FileCopyServer

Inherits:

Object

• Object
• ContentServer::FileCopyServer

Defined in:

lib/content_server/queue_copy.rb

Overview

Simple copier, gets inputs events (files to copy), requests ack from backup to copy then copies one file.

Instance Method Summary

• #initialize(copy_input_queue, port) ⇒ FileCopyServer constructor

  A new instance of FileCopyServer.

• #receive_message(addr_info, message) ⇒ Object
• #run ⇒ Object
• #send_chunk(*arg) ⇒ Object

Constructor Details

#initialize(copy_input_queue, port) ⇒ FileCopyServer

Returns a new instance of FileCopyServer.

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

def initialize(copy_input_queue, port)
  # Local simple tcp connection.
  @backup_tcp = Networking::TCPServer.new(port, method(:receive_message))
  @copy_input_queue = copy_input_queue
  # Stores for each checksum, the file source path.
  # TODO(kolman): If there are items in copy_prepare which timeout (don't get ack),
  # resend the ack request.
  @copy_prepare = {}
  @file_streamer = FileStreamer.new(method(:send_chunk))
  Log.debug3("initialize FileCopyServer on port:%s", port)
  @file_copy_manager = FileCopyManager.new(@copy_input_queue, @file_streamer)
end

Instance Method Details

#receive_message(addr_info, message) ⇒ Object

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

def receive_message(addr_info, message)
  # Add ack message to copy queue.
  Log.debug2("Content server Copy message received: %s", message)
  @copy_input_queue.push(message)
  $process_vars.set('Copy File Queue Size', @copy_input_queue.size)
end

#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

#send_chunk(*arg) ⇒ Object

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

def send_chunk(*arg)
  @copy_input_queue.push([:COPY_CHUNK, arg])
  $process_vars.set('Copy File Queue Size', @copy_input_queue.size)
end