def send_outofband_data(data, restart_pos = 0)
      wait_for_datasocket do |datasocket|
        if datasocket.nil?
          send_response "425 Error establishing connection"
        else
          if data.is_a?(Array)
            data = data.join(LBRK) << LBRK
          end
          data = StringIO.new(data) if data.kind_of?(String)

          data.seek(restart_pos)

          if EM.reactor_running?
            # send the data out in chunks, as fast as the client can recieve it -- not blocking the reactor in the process
            streamer = IOStreamer.new(datasocket, data)
            finalize = Proc.new {
              close_datasocket
              data.close if data.respond_to?(:close) && !data.closed?
            }
            streamer.callback {
              send_response "226 Closing data connection, sent #{streamer.bytes_streamed} bytes"
              finalize.call
            }
            streamer.errback { |ex|
              send_response "425 Error while streaming data, sent #{streamer.bytes_streamed} bytes"
              finalize.call
              raise ex
            }
          else
            # blocks until all data is sent
            begin
              bytes = 0
              data.each do |line|
                datasocket.send_data(line)
                bytes += line.bytesize
              end
              send_response "226 Closing data connection, sent #{bytes} bytes"
            ensure
              close_datasocket
              data.close if data.respond_to?(:close)
            end
          end
        end
      end
    end
def send_outofband_data(data, restart_pos = 0)
      wait_for_datasocket do |datasocket|
        if datasocket.nil?
          send_response "425 Error establishing connection"
        else
          if data.is_a?(Array)
            data = data.join(LBRK) << LBRK
          end
          data = StringIO.new(data) if data.kind_of?(String)

          data.seek(restart_pos)

          if EM.reactor_running?
            # send the data out in chunks, as fast as the client can recieve it -- not blocking the reactor in the process
            streamer = IOStreamer.new(datasocket, data)
            finalize = Proc.new {
              close_datasocket
              data.close if data.respond_to?(:close) && !data.closed?
            }
            streamer.callback {
              send_response "226 Closing data connection, sent #{streamer.bytes_streamed} bytes"
              finalize.call
            }
            streamer.errback { |ex|
              send_response "425 Error while streaming data, sent #{streamer.bytes_streamed} bytes"
              finalize.call
              raise ex
            }
          else
            # blocks until all data is sent
            begin
              bytes = 0
              data.each do |line|
                datasocket.send_data(line)
                bytes += line.bytesize
              end
              send_response "226 Closing data connection, sent #{bytes} bytes"
            ensure
              close_datasocket
              data.close if data.respond_to?(:close)
            end
          end
        end
      end
    end
send data to the client across the data socket.

 The data socket is NOT guaranteed to be setup by the time this method runs.
 If it isn't ready yet, exit the method and try again on the next reactor
 tick. This is particularly likely with some clients that operate in passive
 mode. They get a message on the control port with the data port details, so
 they start up a new data connection AND send they command that will use it
 in close succession.

 The data port setup needs to complete a TCP handshake before it will be
 ready to use, so it may take a few RTTs after the command is received at
 the server before the data socket is ready.