# encoding: utf-8

require 'base64'

module OneApm
  module Support
    module ForkedProcessChannel
      extend self

      def register_report_channel(id)
        listener.register_pipe(id)
      end

      def channels
        listener.pipes
      end

      def listener
        @listener ||= Listener.new
      end

      class Pipe

        READY_MARKER = "READY"
        NUM_LENGTH_BYTES = 4

        attr_accessor :in, :out
        attr_reader :last_read, :parent_pid

        def initialize
          @out, @in = IO.pipe
          if defined?(::Encoding::ASCII_8BIT)
            @in.set_encoding(::Encoding::ASCII_8BIT)
          end
          @last_read = Time.now
          @parent_pid = $$
        end

        def close
          @out.close unless @out.closed?
          @in.close unless @in.closed?
        end

        def serialize_message_length(data)
          [data.bytesize].pack("L>")
        end

        def deserialize_message_length(data)
          data.unpack("L>").first
        end

        def write(data)
          @out.close unless @out.closed?
          @in << serialize_message_length(data)
          @in << data
        end

        def read
          @in.close unless @in.closed?
          @last_read = Time.now
          length_bytes = @out.read(NUM_LENGTH_BYTES)
          if length_bytes
            message_length = deserialize_message_length(length_bytes)
            if message_length
              @out.read(message_length)
            else
              length_hex = length_bytes.bytes.map { |b| b.to_s(16) }.join(' ')
              OneApm::Manager.logger.error("Failed to deserialize message length from pipe. Bytes: [#{length_hex}]")
              nil
            end
          else
            OneApm::Manager.logger.error("Failed to read bytes for length from pipe.")
            nil
          end
        end

        def eof?
          !@out.closed? && @out.eof?
        end

        def after_fork_in_child
          @out.close unless @out.closed?
          write(READY_MARKER)
        end

        def after_fork_in_parent
          @in.close unless @in.closed?
        end

        def closed?
          @out.closed? && @in.closed?
        end

      end

      class Listener

        attr_reader   :thread
        attr_accessor :pipes, :timeout, :select_timeout

        def initialize
          @pipes = {}
          @pipes_lock = Mutex.new

          @timeout = 360
          @select_timeout = 60
        end

        def wakeup
          wake.in << '.'
        end

        def register_pipe(id)
          @pipes_lock.synchronize do
            @pipes[id] = Pipe.new
          end

          wakeup
        end

        def start
          return if @started == true
          @started = true
          @thread = OneApm::Agent::Threading::AgentThread.create('Pipe Channel Manager') do
            now = nil
            loop do
              clean_up_pipes

              pipes_to_listen_to = @pipes_lock.synchronize do
                @pipes.values.map{|pipe| pipe.out} + [wake.out]
              end

              OneApm::Manager.record_metric('Supportability/Listeners', (Time.now - now).to_f) if now

              if ready = IO.select(pipes_to_listen_to, [], [], @select_timeout)
                now = Time.now

                ready_pipes = ready[0]
                ready_pipes.each do |pipe|
                  merge_data_from_pipe(pipe) unless pipe == wake.out
                end

                wake.out.read(1) if ready_pipes.include?(wake.out)
              end

              break unless should_keep_listening?
            end
          end
          sleep 0.001
        end

        def stop_listener_thread
          @started = false
          wakeup
          @thread.join
        end

        def stop
          return unless @started == true
          stop_listener_thread
          close_all_pipes
          @wake.close
          @wake = nil
        end

        def close_all_pipes
          @pipes_lock.synchronize do
            @pipes.each do |id, pipe|
              pipe.close if pipe
            end
            @pipes = {}
          end
        end

        def wake
          @wake ||= Pipe.new
        end

        def started?
          @started
        end

        protected

        def merge_data_from_pipe(pipe_handle)
          pipe = find_pipe_for_handle(pipe_handle)
          raw_payload = pipe.read
          if raw_payload && !raw_payload.empty?
            if raw_payload == Pipe::READY_MARKER
              pipe.after_fork_in_parent
            else
              payload = unmarshal(raw_payload)
              if payload
                endpoint, items = payload
                OneApm::Manager.agent.merge_data_for_endpoint(endpoint, items)
              end
            end
          end

          pipe.close if pipe.eof?
        end

        def unmarshal(data)
          OneApm::LanguageSupport.with_cautious_gc do
            Marshal.load(data)
          end
        rescue StandardError => e
          OneApm::Manager.logger.error "Failure unmarshalling message from Resque child process", e
          OneApm::Manager.logger.debug Base64.encode64(data)
          nil
        end

        def should_keep_listening?
          @started || @pipes_lock.synchronize { @pipes.values.find{|pipe| !pipe.in.closed?} }
        end

        def clean_up_pipes
          @pipes_lock.synchronize do
            @pipes.values.each do |pipe|
              if pipe.last_read.to_f + @timeout < Time.now.to_f
                pipe.close unless pipe.closed?
              end
            end
            @pipes.reject! {|id, pipe| pipe.out.closed? }
          end
        end

        def find_pipe_for_handle(out_handle)
          @pipes_lock.synchronize do
            @pipes.values.find{|pipe| pipe.out == out_handle }
          end
        end

      end
    end
  end
end