http: ignore upgrade header when no upgrade listener is present

deps/http.lua

@@ -160,7 +160,6 @@ function ServerResponse:flushHeaders()
   head.code = statusCode
   local h = self.encode(head)
   self.socket:write(h)
-
 end
 
 function ServerResponse:write(chunk, callback)
@@ -346,12 +345,18 @@ function ClientRequest:initialize(options, callback)
   self.encode = codec.encoder()
   self.decode = codec.decoder()
 
+  if callback then
+    self:once('response', callback)
+  end
+
   local buffer = ''
   local res
 
   local function flush()
-    res:push()
-    res = nil
+    if res then
+      res:push()
+      res = nil
+    end
   end
 
   local socket = options.socket or net.createConnection(self.port, self.host)
@@ -363,51 +368,37 @@ function ClientRequest:initialize(options, callback)
     self.connected = true
     self:emit('socket', socket)
 
-    local function onEnd()
-      -- Just in case the stream ended and we still had an open response,
-      -- end it.
-      if res then flush() end
-    end
     local function onData(chunk)
       -- Run the chunk through the decoder by concatenating and looping
       buffer = buffer .. chunk
       while true do
         local R, event, extra = pcall(self.decode,buffer)
         if R==true then
           -- nil extra means the decoder needs more data, we're done here.
-          if not extra then break end
+          if not extra then return end
           -- Store the leftover data.
           buffer = extra
           if type(event) == "table" then
-            if self.method ~= 'CONNECT' or res == nil then
-              -- If there was an old response that never closed, end it.
-              if res then flush() end
-              -- Create a new response object
+            if not res then
+              flush()
               res = IncomingMessage:new(event, socket)
-              -- If the request upgrades the protocol then detatch the listeners so http codec is no longer used
-              local is_upgraded
-              if res.headers.upgrade then
-                is_upgraded = true
-                socket:removeListener("data", onData)
-                socket:removeListener("end", onEnd)
+            end
+            if self.method == 'CONNECT' or res.headers.upgrade then
+              local evt = self.method == 'CONNECT' and 'connect' or 'upgrade'
+              if self:listenerCount(evt) > 0 then
+                socket:removeListener('data', onData)
+                socket:removeListener('end', flush)
                 socket:read(0)
                 if #buffer > 0 then
                   socket:pause()
                   socket:unshift(buffer)
                 end
-              end
-              -- Call the user callback to handle the response
-              if callback then
-                callback(res)
-              end
-              self:emit('response', res)
-              if is_upgraded then
-                break
+                return self:emit(evt, res, socket, event)
+              elseif self.method == 'CONNECT' or res.statusCode == 101 then
+                return self:destroy()
               end
             end
-            if self.method == 'CONNECT' then
-              self:emit('connect', res, socket, event)
-            end
+            self:emit('response', res)
           elseif res and type(event) == "string" then
             if #event == 0 then
               -- Empty string in http-decoder means end of body
@@ -423,18 +414,16 @@ function ClientRequest:initialize(options, callback)
             end
           end
         else
-          self:emit('error', event)
-          break
+          return self:emit('error', event)
         end
       end
     end
     socket:on('data', onData)
-    socket:on('end', onEnd)
+    socket:on('end', flush)
 
     if self.ended then
-      self:_done(self.ended.data, self.ended.cb)
+      return self:_done(self.ended.data, self.ended.cb)
     end
-
   end)
 end
 

tests/test-http-client.lua

@@ -35,12 +35,13 @@ require('tap')(function(test)
   end)
 
   test("http-client (errors are bubbled)", function(expect)
-    local socket = http.get('http://github.com:1234', function (res)
+    local req = http.get('http://unknown123456789.com:1234', function (res)
       assert(false)
     end)
-    socket:on('error',expect(function(err)
+    req:on('error',expect(function(err)
       assert(not (err == nil))
     end))
+    req:setTimeout(25000)
   end)
 
   test("http-client stream file", function(expect)
@@ -69,6 +70,134 @@ require('tap')(function(test)
       end)
     end)
   end)
+
+  test("http-client ignore HTTP/2 upgrade from server", function(expect)
+    local port = 50020
+
+    local server
+    server = http.createServer(expect(function(req, res)
+      print('server request', req.method)
+      assert(req.method == 'GET')
+      assert(req.url == 'test_path')
+      -- advertise HTTP/2 support
+      res:setHeader('Upgrade', 'h2,h2c')
+      return res:finish('response')
+    end))
+
+    server:listen(port, expect(function()
+      print('server running')
+      local options = {
+        port = port,
+        path = 'test_path',
+      }
+
+      local req = http.request(options, function(res)
+        print('req cb')
+        assert(res.headers.upgrade == 'h2,h2c')
+        res:on('data', function(data)
+          print('req data')
+          assert(data == 'response')
+          server:destroy()
+        end)
+      end)
+
+      req:done()
+    end))
+  end)
+
+  test("http-client accept Websocket upgrade from server", function(expect)
+    local port = 50030
+
+    local server
+    server = http.createServer(expect(function(req, res)
+      print('server request', req.method)
+      assert(req.method == 'GET')
+      assert(req.url == 'test_path')
+      assert(req.headers.upgrade == 'websocket')
+      res:writeHead(101, {
+        ['Upgrade'] = 'websocket',
+        ['Connection'] = 'Upgrade',
+        -- other ws headers
+      })
+      res:flushHeaders()
+
+      res.socket:on('data', function(data)
+        print('response data')
+        assert(data == 'request')
+        res.socket:destroy()
+        server:destroy()
+      end)
+
+      res.socket:write('response')
+    end))
+
+    server:listen(port, expect(function()
+      print('server running')
+      local options = {
+        port = port,
+        path = 'test_path',
+        headers = {
+          ['Upgrade'] = 'websocket',
+          ['Connection'] = 'Upgrade',
+          -- other ws headers
+        }
+      }
+
+      local req = http.request(options, function(res)
+        error('response callback should not be called')
+      end)
+
+      req:once('upgrade', expect(function(res, socket, event)
+        print('req upgrade')
+        assert(res.headers.upgrade == 'websocket')
+        socket:resume()
+        socket:write('request')
+      end))
+
+      req:done()
+    end))
+  end)
+
+  test("http-client connect", function(expect)
+    local port = 50040
+
+    local server
+    server = http.createServer(expect(function(req, res)
+      print('server request', req.method)
+      assert(req.method == 'CONNECT')
+      assert(req.url == 'test_path')
+      -- TODO: CONNECT should be detached from http codec.
+      return res:finish('response')
+    end))
+
+    server:listen(port, expect(function()
+      print('server running')
+      local options = {
+        port = port,
+        method = 'CONNECT',
+        path = 'test_path',
+      }
+
+      local req = http.request(options, function(res)
+        error('response callback should not be called')
+      end)
+
+      req:once('connect', expect(function(res, socket, event)
+        print('req connect')
+        socket:on('data', function(data)
+          print('socket data')
+          assert(data)
+          assert(data:find('response'))
+          server:destroy()
+        end)
+
+        socket:resume()
+      end))
+
+      req:done()
+    end))
+  end)
+
 end)