rack: add fallback-legacy option (fallback to backend HTTP/1.x server if client doesn't support HTTP/2)

7 files changed, 102 insertions, 42 deletions

diff --git a/lib/plum/errors.rb b/lib/plum/errors.rb
index 09b5809..3b1b2fc 100644
--- a/lib/plum/errors.rb
+++ b/lib/plum/errors.rb
@@ -44,12 +44,11 @@ module Plum
   class LocalStreamError < LocalHTTPError; end
 
   class LegacyHTTPError < Error
-    attr_reader :headers, :data, :parser
+    attr_reader :buf
 
-    def initialize(headers, data, parser)
-      @headers = headers
-      @data = data
-      @parser = parser
+    def initialize(message, buf = nil)
+      super(message)
+      @buf = buf
     end
   end
 
diff --git a/lib/plum/rack/cli.rb b/lib/plum/rack/cli.rb
index a195db8..635a138 100644
--- a/lib/plum/rack/cli.rb
+++ b/lib/plum/rack/cli.rb
@@ -46,6 +46,12 @@ module Plum
         config[:server_push] = @options[:server_push] unless @options[:server_push].nil?
         config[:threaded] = @options[:threaded] unless @options[:threaded].nil?
 
+        if @options[:fallback_legacy]
+          h, p = @options[:fallback_legacy].split(":")
+          config[:fallback_legacy_host] = h
+          config[:fallback_legacy_port] = p.to_i
+        end
+
         if @options[:socket]
           config[:listeners] << { listener: UNIXListener,
                                   path: @options[:socket] }
@@ -118,6 +124,10 @@ module Plum
             @options[:threaded] = true
           end
 
+          o.on "--fallback-legacy HOST:PORT", "Fallbacks if the client doesn't support HTTP/2" do |arg|
+            @options[:fallback_legacy] = arg
+          end
+
           o.on "-v", "--version", "Show version" do
             puts "plum version #{::Plum::VERSION}"
             exit(0)
diff --git a/lib/plum/rack/dsl.rb b/lib/plum/rack/dsl.rb
index 6e8dda6..b1f8bb0 100644
--- a/lib/plum/rack/dsl.rb
+++ b/lib/plum/rack/dsl.rb
@@ -43,6 +43,12 @@ module Plum
         def threaded(bool)
           @config[:threaded] = !!bool
         end
+
+        def fallback_legacy(str)
+          h, p = str.split(":")
+          @config[:fallback_legacy_host] = h
+          @config[:fallback_legacy_port] = p.to_i
+        end
       end
     end
   end
diff --git a/lib/plum/rack/listener.rb b/lib/plum/rack/listener.rb
index 6543e7c..079c680 100644
--- a/lib/plum/rack/listener.rb
+++ b/lib/plum/rack/listener.rb
@@ -42,8 +42,11 @@ module Plum
         ctx = OpenSSL::SSL::SSLContext.new
         ctx.ssl_version = :TLSv1_2
         ctx.alpn_select_cb = -> protocols {
-          raise "Client does not support HTTP/2: #{protocols}" unless protocols.include?("h2")
-          "h2"
+          if protocols.include?("h2")
+            "h2"
+          else
+            protocols.first
+          end
         }
         ctx.tmp_ecdh_callback = -> (sock, ise, keyl) { OpenSSL::PKey::EC.new("prime256v1") }
         ctx.cert = OpenSSL::X509::Certificate.new(cert)
@@ -58,6 +61,7 @@ module Plum
       end
 
       def plum(sock)
+        raise ::Plum::LegacyHTTPError.new("client doesn't offered h2 with ALPN", nil) unless sock.alpn_protocol == "h2"
         ::Plum::HTTPSServerConnection.new(sock)
       end
 
diff --git a/lib/plum/rack/server.rb b/lib/plum/rack/server.rb
index 26ead2a..1f3d32c 100644
--- a/lib/plum/rack/server.rb
+++ b/lib/plum/rack/server.rb
@@ -47,31 +47,59 @@ module Plum
         sock = svr.accept
         Thread.new {
           begin
-            sock = sock.accept if sock.respond_to?(:accept)
-            plum = svr.plum(sock)
+            begin
+              sock = sock.accept if sock.respond_to?(:accept)
+              plum = svr.plum(sock)
 
-            con = Session.new(app: @app,
-                              plum: plum,
-                              sock: sock,
-                              logger: @logger,
-                              config: @config,
-                              remote_addr: sock.peeraddr.last)
-            con.run
-          rescue Errno::ECONNRESET, Errno::ECONNABORTED, Errno::EPROTO, Errno::EINVAL => e # closed
-            sock.close if sock
+              con = Session.new(app: @app,
+                                plum: plum,
+                                sock: sock,
+                                logger: @logger,
+                                config: @config,
+                                remote_addr: sock.peeraddr.last)
+              con.run
+            rescue ::Plum::LegacyHTTPError => e
+              @logger.info "legacy HTTP client: #{e}"
+              handle_legacy(e, sock)
+            end
+          rescue Errno::ECONNRESET, Errno::EPROTO, Errno::EINVAL, EOFError => e # closed
           rescue StandardError => e
             log_exception(e)
+          ensure
             sock.close if sock
           end
         }
       rescue Errno::ECONNRESET, Errno::ECONNABORTED, Errno::EPROTO, Errno::EINVAL => e # closed
+        sock.close if sock
       rescue StandardError => e
         log_exception(e)
+        sock.close if sock
       end
 
       def log_exception(e)
         @logger.error("#{e.class}: #{e.message}\n#{e.backtrace.map { |b| "\t#{b}" }.join("\n")}")
       end
+
+      def handle_legacy(e, sock)
+        if @config[:fallback_legacy_host]
+          @logger.info "legacy HTTP: fallbacking to: #{@config[:fallback_legacy_host]}:#{@config[:fallback_legacy_port]}"
+          upstream = TCPSocket.open(@config[:fallback_legacy_host], @config[:fallback_legacy_port])
+          upstream.write(e.buf) if e.buf
+          loop do
+            ret = IO.select([sock, upstream])
+            ret[0].each { |s|
+              a = s.readpartial(65536)
+              if s == upstream
+                sock.write(a)
+              else
+                upstream.write(a)
+              end
+            }
+          end
+        end
+      ensure
+        upstream.close if upstream
+      end
     end
   end
 end
diff --git a/lib/plum/rack/session.rb b/lib/plum/rack/session.rb
index d13c86d..76e91c0 100644
--- a/lib/plum/rack/session.rb
+++ b/lib/plum/rack/session.rb
@@ -38,10 +38,20 @@ module Plum
 
       private
       def setup_plum
+        @plum.on(:frame) { |f| puts "recv:#{f.inspect}" }
+        @plum.on(:send_frame) { |f|
+          puts "send:#{f.inspect}" unless f.type == :data
+        }
         @plum.on(:connection_error) { |ex| @logger.error(ex) }
 
         # @plum.on(:stream) { |stream| @logger.debug("new stream: #{stream}") }
-        @plum.on(:stream_error) { |stream, ex| @logger.error(ex) }
+        @plum.on(:stream_error) { |stream, ex|
+          if [:cancel, :refused_stream].include?(ex.http2_error_type)
+            @logger.debug("stream was cancelled: #{stream}")
+          else
+            @logger.error(ex)
+          end
+        }
 
         reqs = {}
         @plum.on(:headers) { |stream, h|
diff --git a/lib/plum/server/http_connection.rb b/lib/plum/server/http_connection.rb
index 8f91ca4..1860126 100644
--- a/lib/plum/server/http_connection.rb
+++ b/lib/plum/server/http_connection.rb
@@ -7,10 +7,9 @@ module Plum
 
     def initialize(sock, local_settings = {})
       require "http/parser"
-      @_headers = nil
-      @_body = String.new
-      @_http_parser = setup_parser
       @sock = sock
+      @negobuf = String.new
+      @_http_parser = setup_parser
       super(@sock.method(:write), local_settings)
     end
 
@@ -23,40 +22,44 @@ module Plum
     private
     def negotiate!
       super
-    rescue RemoteConnectionError
-      # Upgrade from HTTP/1.1
+    rescue RemoteConnectionError # Upgrade from HTTP/1.1 or legacy
+      @negobuf << @buffer
       offset = @_http_parser << @buffer
       @buffer.byteshift(offset)
     end
 
     def setup_parser
+      headers = nil
+      body = String.new
+
       parser = HTTP::Parser.new
-      parser.on_headers_complete = proc {|_headers|
-        @_headers = _headers.map {|n, v| [n.downcase, v] }.to_h
+      parser.on_headers_complete = proc { |_headers|
+        headers = _headers.map {|n, v| [n.downcase, v] }.to_h
       }
-      parser.on_body = proc {|chunk| @_body << chunk }
-      parser.on_message_complete = proc {|env|
-        connection = @_headers["connection"] || ""
-        upgrade = @_headers["upgrade"] || ""
-        settings = @_headers["http2-settings"]
+      parser.on_body = proc { |chunk| body << chunk }
+      parser.on_message_complete = proc { |env|
+        connection = headers["connection"] || ""
+        upgrade = headers["upgrade"] || ""
+        settings = headers["http2-settings"]
 
         if (connection.split(", ").sort == ["Upgrade", "HTTP2-Settings"].sort &&
             upgrade.split(", ").include?("h2c") &&
             settings != nil)
-          switch_protocol(settings)
+          switch_protocol(settings, parser, headers, body)
+          @negobuf = @_http_parser = nil
         else
-          raise LegacyHTTPError.new(@_headers, @_body, parser)
+          raise LegacyHTTPError.new("request doesn't Upgrade", @negobuf)
         end
       }
 
       parser
     end
 
-    def switch_protocol(settings)
+    def switch_protocol(settings, parser, headers, data)
       self.on(:negotiated) {
         _frame = Frame.new(type: :settings, stream_id: 0, payload: Base64.urlsafe_decode64(settings))
         receive_settings(_frame, send_ack: false) # HTTP2-Settings
-        process_first_request
+        process_first_request(parser, headers, data)
       }
 
       resp = "HTTP/1.1 101 Switching Protocols\r\n" +
@@ -68,17 +71,17 @@ module Plum
       @sock.write(resp)
     end
 
-    def process_first_request
+    def process_first_request(parser, headers, body)
       encoder = HPACK::Encoder.new(0, indexing: false) # don't pollute connection's HPACK context
       stream = stream(1)
       max_frame_size = local_settings[:max_frame_size]
-      headers = @_headers.merge({ ":method" => @_http_parser.http_method,
-                                  ":path" => @_http_parser.request_url,
-                                  ":authority" => @_headers["host"] })
-                         .reject {|n, v| ["connection", "http2-settings", "upgrade", "host"].include?(n) }
+      nheaders = headers.merge({ ":method" => parser.http_method,
+                                 ":path" => parser.request_url,
+                                 ":authority" => headers["host"] })
+                        .reject {|n, v| ["connection", "http2-settings", "upgrade", "host"].include?(n) }
 
-      stream.receive_frame Frame.headers(1, encoder.encode(headers), end_headers: true)
-      stream.receive_frame Frame.data(1, @_body, end_stream: true)
+      stream.receive_frame Frame.headers(1, encoder.encode(nheaders), end_headers: true)
+      stream.receive_frame Frame.data(1, body, end_stream: true)
     end
   end
 end