# frozen_string_literal: true require "json" module Vamp # Owns the subprocess wire protocol, isolated from user stdout and callbacks. class Protocol # @return [void] def initialize # Capture physical streams; globals may be replaced by user code. @output = STDOUT.dup # rubocop:disable Style/GlobalStdStream @output.binmode @output.sync = true @input = STDIN.dup # rubocop:disable Style/GlobalStdStream @input.binmode @mutex = Mutex.new # Explicit STDOUT.write must also stay off the duplicated protocol pipe. STDOUT.reopen(STDERR) # rubocop:disable Style/GlobalStdStream $stdout = STDERR # rubocop:disable Style/GlobalStdStream Thread.abort_on_exception = true end # @param arguments [Array] mode, source path, and startup JSON # @return [Object] the shutdown callback's result def run(arguments) start(arguments) send_frame(Vamp.startup) process_requests if @mode == "plugin" Vamp.shutdown ensure # Includes SystemExit/Interrupt, without rescuing or hiding callback errors. error = $! # rubocop:disable Style/SpecialGlobalVars -- Native exception state at the process boundary. report_error(error) if error end private # @param arguments [Array] # @return [Boolean] the source load result def start(arguments) @mode, path, encoded_startup = arguments valid = %w[config plugin].include?(@mode) && path && encoded_startup raise ArgumentError, "expected config or plugin mode, path, and startup data" unless valid raise ArgumentError, "startup frame too large" if encoded_startup.bytesize > MAX_FRAME Vamp.setup(@mode, JSON.parse(encoded_startup), method(:send_frame)) load path end # @param frame [Hash{String => Object}] # @return [Integer] bytes written by the terminating newline def send_frame(frame) @mutex.synchronize do payload = JSON.generate(frame) raise ArgumentError, "outgoing frame too large" if payload.bytesize > MAX_FRAME @output.write(payload) @output.write("\n") end end # @return [void] def process_requests loop do @request_id = nil line = @input.gets(MAX_FRAME + 2) break if line.nil? request = parse_request(line) remember_request(request) dispatch(request) break if request["type"] == "shutdown" end end # @param line [String] # @return [Object] parsed JSON, validated by the selected request handler def parse_request(line) unless line.end_with?("\n") && line.bytesize - 1 <= MAX_FRAME raise ArgumentError, "invalid or oversized protocol frame" end JSON.parse(line) end # @param request [Object] # @return [Integer, nil] ID to attach to a failed invocation def remember_request(request) return unless request.is_a?(Hash) id = request["id"] @request_id = id if id.is_a?(Integer) && id.positive? end # @param request [Object] # @return [Object] def dispatch(request) case request.is_a?(Hash) && request["type"] when "invoke" send_frame(Vamp.invoke(request)) when "document" Vamp.document(request) when "shutdown" validate_shutdown(request) else raise ArgumentError, "unknown protocol frame" end end # @param request [Hash{String => Object}] # @return [nil] def validate_shutdown(request) return if request.keys.sort == %w[type version] && request["version"].is_a?(Integer) && request["version"] == 1 raise ArgumentError, "invalid shutdown frame" end # @param error [Exception] # @return [void] def report_error(error) # No callback effects survive an exception. Bound and sanitize the message. message = "#{error.class}: #{error.message}".encode("UTF-8", invalid: :replace, undef: :replace) frame = { "version" => 1, "type" => "error", "message" => message.each_char.take(2048).join } frame["id"] = @request_id unless @request_id.nil? send_frame(frame) ensure # If even reporting fails, Rust observes EOF and invalidates the host. exit 1 end end end Vamp::Protocol.new.run(ARGV)