Class: ContentServer::FileCopyServer

Inherits:
Object
  • Object
show all
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 collapse

Constructor Details

#initialize(copy_input_queue, port) ⇒ FileCopyServer

Returns a new instance of FileCopyServer.



26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/content_server/queue_copy.rb', line 26

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:#{port}")
end

Instance Method Details

#receive_message(addr_info, message) ⇒ Object 



42
43
44
45
46
 
# File 'lib/content_server/queue_copy.rb', line 42

def receive_message(addr_info, message)
  # Add ack message to copy queue.
  Log.debug2("Content server Copy message received: #{message}")
  @copy_input_queue.push(message)
end

#run ⇒ Object 



48
49
50
51
52
53
54
55
56
57
58
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
94
95
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
 
# File 'lib/content_server/queue_copy.rb', line 48

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

      if message_type == :COPY_MESSAGE
        Log.debug1 "Copy files event: #{message_content}"
        # Prepare source,dest map for copy.
        message_content.each_instance { |checksum, size, content_mod_time, instance_mod_time, server, path|
          if !@copy_prepare.key?(checksum) || !@copy_prepare[checksum][1]
            @copy_prepare[checksum] = [path, false]
            Log.debug1("Sending ack for: #{checksum}")
            @backup_tcp.send_obj([:ACK_MESSAGE, [checksum, Time.now.to_i]])
          end
        }
      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}")

        # 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'])
            if !@copy_prepare.key?(checksum) || @copy_prepare[checksum][1]
              Log.warning("File was aborted, copied, or started copy just now: #{checksum}")
            else
              path = @copy_prepare[checksum][0]
              Log.info "Streaming to backup server. content: #{checksum} path:#{path}."
              @file_streamer.start_streaming(checksum, path)
              # Ack received, setting prepare to true
              @copy_prepare[checksum][1] = true
            end
          else
            Log.debug1("Ack timed out span: #{Time.now.to_i - timestamp} > " \
                       "timeout: #{Params['ack_timeout']}")
          end
        else
          Log.debug1('Ack is false');
        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 #{file_checksum}, offset: #{offset} " \
                     "filesize: #{file_size}.")
        # Blocking send.
        @backup_tcp.send_obj([:COPY_CHUNK, message_content])
        if content.nil? and content_checksum.nil?
          @copy_prepare.delete(file_checksum)
        end
      elsif message_type == :ABORT_COPY
        Log.debug1("Aborting file copy: #{message_content}")
        if @copy_prepare.key?(message_content)
          Log.debug1("Aborting: #{@copy_prepare[message_content][0]}")
          @copy_prepare.delete(message_content)
        end
        @file_streamer.abort_streaming(message_content)
      elsif message_type == :RESET_RESUME_COPY
        file_checksum, new_offset = message_content
        Log.debug1("Resetting/Resuming file (#{file_checksum}) copy to #{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 



38
39
40
 
# File 'lib/content_server/queue_copy.rb', line 38

def send_chunk(*arg)
  @copy_input_queue.push([:COPY_CHUNK, arg])
end