Fix upgrade test; correct/add APIs

CHANGELOG.md

@@ -6,13 +6,14 @@ Notable changes to this project are documented in this file. The format is based
 
 Breaking changes:
 - Update node libraries to latest releases (#48 by @JordanMartinez)
-- Reimplement `http`/`https` bindings (#49 by @JordanMartinez)
+- Reimplement `http`/`https` bindings (#49, #50 by @JordanMartinez)
 
 New features:
 
 Bugfixes:
 
 Other improvements:
+- Fix flaky `upgrade` test (#50 by @JordanMartinez)
 
 ## [v8.0.0](https://github.com/purescript-node/purescript-node-http/releases/tag/v8.0.0) - 2022-04-29
 

src/Node/HTTP/ClientRequest.purs

@@ -1,5 +1,6 @@
 module Node.HTTP.ClientRequest
   ( toOutgoingMessage
+  , closeH
   , connectH
   , continueH
   , finishH
@@ -29,13 +30,16 @@ import Node.Buffer (Buffer)
 import Node.EventEmitter (EventHandle(..))
 import Node.EventEmitter.UtilTypes (EventHandle0, EventHandle3, EventHandle1)
 import Node.HTTP.Types (ClientRequest, IMClientRequest, IncomingMessage, OutgoingMessage)
-import Node.Stream (Duplex)
+import Node.Net.Types (Socket, TCP)
 import Unsafe.Coerce (unsafeCoerce)
 
 toOutgoingMessage :: ClientRequest -> OutgoingMessage
 toOutgoingMessage = unsafeCoerce
 
-connectH :: EventHandle3 ClientRequest (IncomingMessage IMClientRequest) Duplex Buffer
+closeH :: EventHandle0 ClientRequest
+closeH = EventHandle "close" identity
+
+connectH :: EventHandle3 ClientRequest (IncomingMessage IMClientRequest) (Socket TCP) Buffer
 connectH = EventHandle "connect" \cb -> mkEffectFn3 \a b c -> cb a b c
 
 continueH :: EventHandle0 ClientRequest
@@ -59,13 +63,13 @@ informationH = EventHandle "information" mkEffectFn1
 responseH :: EventHandle1 ClientRequest (IncomingMessage IMClientRequest)
 responseH = EventHandle "response" mkEffectFn1
 
-socketH :: EventHandle1 ClientRequest Duplex
+socketH :: EventHandle1 ClientRequest (Socket TCP)
 socketH = EventHandle "socket" mkEffectFn1
 
 timeoutH :: EventHandle0 ClientRequest
 timeoutH = EventHandle "timeout" identity
 
-upgradeH :: EventHandle3 ClientRequest (IncomingMessage IMClientRequest) Duplex Buffer
+upgradeH :: EventHandle3 ClientRequest (IncomingMessage IMClientRequest) (Socket TCP) Buffer
 upgradeH = EventHandle "upgrade" \cb -> mkEffectFn3 \a b c -> cb a b c
 
 foreign import path :: ClientRequest -> String

src/Node/HTTP/IncomingMessage.purs

@@ -30,7 +30,8 @@ import Foreign.Object as Object
 import Node.EventEmitter (EventHandle(..))
 import Node.EventEmitter.UtilTypes (EventHandle0)
 import Node.HTTP.Types (IMClientRequest, IMServer, IncomingMessage)
-import Node.Stream (Readable, Duplex)
+import Node.Net.Types (Socket, TCP)
+import Node.Stream (Readable)
 import Unsafe.Coerce (unsafeCoerce)
 
 toReadable :: forall messageType. IncomingMessage messageType -> Readable ()
@@ -65,10 +66,10 @@ rawTrailers im = toMaybe $ rawTrailersImpl im
 
 foreign import rawTrailersImpl :: forall messageType. IncomingMessage messageType -> (Nullable (Array String))
 
-socket :: forall messageType. IncomingMessage messageType -> Effect (Maybe Duplex)
+socket :: forall messageType. IncomingMessage messageType -> Effect (Maybe (Socket TCP))
 socket im = map toMaybe $ runEffectFn1 socketImpl im
 
-foreign import socketImpl :: forall messageType. EffectFn1 (IncomingMessage messageType) (Nullable Duplex)
+foreign import socketImpl :: forall messageType. EffectFn1 (IncomingMessage messageType) (Nullable (Socket TCP))
 
 foreign import statusCode :: IncomingMessage IMClientRequest -> Int
 

src/Node/HTTP/OutgoingMessage.purs

@@ -31,6 +31,7 @@ import Foreign.Object (Object)
 import Node.EventEmitter (EventHandle(..))
 import Node.EventEmitter.UtilTypes (EventHandle0)
 import Node.HTTP.Types (OutgoingMessage)
+import Node.Net.Types (Socket, TCP)
 import Node.Stream (Writable)
 import Unsafe.Coerce (unsafeCoerce)
 
@@ -111,7 +112,7 @@ setTimeout msecs msg = runEffectFn2 setTimeoutImpl msecs msg
 
 foreign import setTimeoutImpl :: EffectFn2 (Milliseconds) (OutgoingMessage) (Unit)
 
-socket :: OutgoingMessage -> Effect (Maybe (Writable ()))
+socket :: OutgoingMessage -> Effect (Maybe (Socket TCP))
 socket msg = map toMaybe $ runEffectFn1 socketImpl msg
 
-foreign import socketImpl :: EffectFn1 (OutgoingMessage) (Nullable (Writable ()))
+foreign import socketImpl :: EffectFn1 (OutgoingMessage) (Nullable (Socket TCP))

src/Node/HTTP/Server.purs

@@ -45,7 +45,7 @@ import Node.Buffer (Buffer)
 import Node.EventEmitter (EventHandle(..))
 import Node.EventEmitter.UtilTypes (EventHandle0, EventHandle2, EventHandle3, EventHandle1)
 import Node.HTTP.Types (Encrypted, HttpServer', IMServer, IncomingMessage, ServerResponse)
-import Node.Net.Types (Server, TCP)
+import Node.Net.Types (Server, Socket, TCP)
 import Node.Stream (Duplex)
 import Node.TLS.Types (TlsServer)
 import Unsafe.Coerce (unsafeCoerce)
@@ -76,7 +76,7 @@ clientErrorH = EventHandle "clientError" \cb -> mkEffectFn2 \a b -> cb a b
 closeH :: forall transmissionType. EventHandle0 (HttpServer' transmissionType)
 closeH = EventHandle "close" identity
 
-connectH :: forall transmissionType. EventHandle3 (HttpServer' transmissionType) (IncomingMessage IMServer) Duplex Buffer
+connectH :: forall transmissionType. EventHandle3 (HttpServer' transmissionType) (IncomingMessage IMServer) (Socket TCP) Buffer
 connectH = EventHandle "connect" \cb -> mkEffectFn3 \a b c -> cb a b c
 
 connectionH :: forall transmissionType. EventHandle1 (HttpServer' transmissionType) Duplex
@@ -88,7 +88,7 @@ dropRequestH = EventHandle "dropRequest" \cb -> mkEffectFn2 \a b -> cb a b
 requestH :: forall transmissionType. EventHandle2 (HttpServer' transmissionType) (IncomingMessage IMServer) ServerResponse
 requestH = EventHandle "request" \cb -> mkEffectFn2 \a b -> cb a b
 
-upgradeH :: forall transmissionType. EventHandle3 (HttpServer' transmissionType) (IncomingMessage IMServer) Duplex Buffer
+upgradeH :: forall transmissionType. EventHandle3 (HttpServer' transmissionType) (IncomingMessage IMServer) (Socket TCP) Buffer
 upgradeH = EventHandle "upgrade" \cb -> mkEffectFn3 \a b c -> cb a b c
 
 closeAllConnections :: forall transmissionType. (HttpServer' transmissionType) -> Effect Unit

test/Main.purs

@@ -2,16 +2,18 @@ module Test.Main where
 
 import Prelude
 
-import Data.Foldable (foldMap)
+import Data.Either (Either(..))
+import Data.Foldable (foldMap, for_)
 import Data.Maybe (fromMaybe)
 import Effect (Effect)
+import Effect.Aff (launchAff_, makeAff, nonCanceler)
+import Effect.Class (liftEffect)
 import Effect.Console (log, logShow)
 import Effect.Uncurried (EffectFn2)
 import Foreign.Object (lookup)
-import Node.Buffer (Buffer)
 import Node.Buffer as Buffer
 import Node.Encoding (Encoding(..))
-import Node.EventEmitter (once_)
+import Node.EventEmitter (once, once_)
 import Node.HTTP as HTTP
 import Node.HTTP.ClientRequest as Client
 import Node.HTTP.IncomingMessage as IM
@@ -23,7 +25,8 @@ import Node.HTTP.Types (HttpServer', IMServer, IncomingMessage, ServerResponse)
 import Node.HTTPS as HTTPS
 import Node.Net.Server (listenTcp)
 import Node.Net.Server as NetServer
-import Node.Stream (Duplex, Writable, end, pipe)
+import Node.Net.Socket as Socket
+import Node.Stream (Writable, end, pipe)
 import Node.Stream as Stream
 import Partial.Unsafe (unsafeCrashWith)
 import Unsafe.Coerce (unsafeCoerce)
@@ -35,7 +38,7 @@ foreign import stdout :: forall r. Writable r
 main :: Effect Unit
 main = do
   testBasic
-  -- testUpgrade
+  testUpgrade
   testHttpsServer
   testHttps
   testCookies
@@ -195,44 +198,60 @@ logResponse response = void do
   pipe (IM.toReadable response) stdout
 
 testUpgrade :: Effect Unit
-testUpgrade = do
-  server <- HTTP.createServer
-  server # once_ Server.upgradeH handleUpgrade
-
-  server # once_ Server.requestH (respond (mempty))
+testUpgrade = launchAff_ do
+  server <- liftEffect HTTP.createServer
   let netServer = Server.toNetServer server
-  netServer # once_ NetServer.listeningH do
-    log $ "Listening on port " <> show httpPort <> "."
-    sendRequests
-  listenTcp netServer { host: "localhost", port: httpPort }
+  waitUntilListening netServer
+
+  -- This tests that the upgrade callback is not called when the request is not an HTTP upgrade
+  doRegularRequest server
+
+  -- These two requests test that the upgrade callback is called and that it has
+  -- access to the original request and can write to the underlying TCP socket
+  checkUpgradeRequest server
+  checkWebSocketUpgrade server
+
+  liftEffect do
+    closeAllConnections server
+    NetServer.close netServer
   where
   httpPort = 3000
 
-  handleUpgrade :: IncomingMessage IMServer -> Duplex -> Buffer -> Effect Unit
-  handleUpgrade req socket _ = do
-    let upgradeHeader = fromMaybe "" $ lookup "upgrade" $ IM.headers req
-    if upgradeHeader == "websocket" then
-      void $ Stream.writeString socket UTF8
-        "HTTP/1.1 101 Switching Protocols\r\nContent-Length: 0\r\n\r\n"
-    else
-      void $ Stream.writeString socket UTF8
-        "HTTP/1.1 426 Upgrade Required\r\nContent-Length: 0\r\n\r\n"
+  waitUntilListening netServer = makeAff \done -> do
+    netServer # once_ NetServer.listeningH do
+      liftEffect $ log $ "Listening on port " <> show httpPort <> "."
+      done $ Right unit
+    listenTcp netServer { host: "localhost", port: httpPort }
+    pure nonCanceler
+
+  doRegularRequest server = makeAff \done -> do
+    rmListener <- server # once Server.upgradeH \_ _ _ -> do
+      unsafeCrashWith "testUpgrade - regularRequest - got an upgrade request when expected simple request"
+    server # once_ Server.requestH (respond mempty)
 
-  sendRequests :: Effect Unit
-  sendRequests = do
-    -- This tests that the upgrade callback is not called when the request is not an HTTP upgrade
     reqSimple <- HTTP.requestOpts { port: httpPort }
     reqSimple # once_ Client.responseH \response -> do
       if (IM.statusCode response /= 200) then
-        unsafeCrashWith "Unexpected response to simple request on `testUpgrade`"
-      else
-        pure unit
+        unsafeCrashWith $ "testUpgrade - regularRequest - unexpected response to simple request: " <> show (IM.statusCode response)
+      else do
+        rmListener
+        log "testUpgrade - regularRequest - Got regular response."
+        done $ Right unit
     end (OM.toWriteable $ Client.toOutgoingMessage reqSimple)
+    pure nonCanceler
+
+  checkUpgradeRequest server = makeAff \done -> do
+    rmListener <- server # once Server.requestH \_ -> do
+      unsafeCrashWith "testUpgrade - checkUpgradeRequest - request handler fired instead of upgrade handler"
+    server # once_ Server.upgradeH \req socket _ -> do
+      case fromMaybe "" $ lookup "upgrade" $ IM.headers req of
+        "websocket" ->
+          unsafeCrashWith "testUpgrade - checkUpgradeRequest - expected non-websocket upgrade but got websocket upgrade"
+        _ -> do
+          void $ Stream.writeString (Socket.toDuplex socket) UTF8
+            "HTTP/1.1 426 Upgrade Required\r\nContent-Length: 0\r\n\r\n"
+          void $ Stream.end (Socket.toDuplex socket)
 
-    {-
-      These two requests test that the upgrade callback is called and that it has
-      access to the original request and can write to the underlying TCP socket
-    -}
     reqUpgrade <- HTTP.requestOpts
       { port: httpPort
       , headers: unsafeCoerce
@@ -242,10 +261,25 @@ testUpgrade = do
       }
     reqUpgrade # once_ Client.responseH \response -> do
       if (IM.statusCode response /= 426) then
-        unsafeCrashWith "Unexpected response to upgrade request on `testUpgrade`"
-      else
-        pure unit
+        unsafeCrashWith $ "Unexpected response to upgrade request on `testUpgrade`: " <> show (IM.statusCode response)
+      else do
+        rmListener
+        log "testUpgrade - checkUpgradeRequest - Got upgrade required response."
+        done $ Right unit
     end (OM.toWriteable $ Client.toOutgoingMessage reqUpgrade)
+    pure nonCanceler
+
+  checkWebSocketUpgrade server = makeAff \done -> do
+    rmListener <- server # once Server.requestH \_ -> do
+      unsafeCrashWith "testUpgrade - checkWebSocketUpgrade - request handler fired instead of upgrade handler"
+    server # once_ Server.upgradeH \req socket _ -> do
+      case fromMaybe "" $ lookup "upgrade" $ IM.headers req of
+        "websocket" -> do
+          void $ Stream.writeString (Socket.toDuplex socket) UTF8
+            "HTTP/1.1 101 Switching Protocols\r\nContent-Length: 0\r\n\r\n"
+          void $ Stream.end (Socket.toDuplex socket)
+        _ ->
+          unsafeCrashWith "testUpgrade - checkWebSocketUpgrade - expected websocket upgrade but got non-websocket upgrade"
 
     reqWSUpgrade <- HTTP.requestOpts
       { port: httpPort
@@ -257,6 +291,15 @@ testUpgrade = do
     reqWSUpgrade # once_ Client.responseH \response -> do
       if (IM.statusCode response /= 101) then
         unsafeCrashWith "Unexpected response to websocket upgrade request on `testUpgrade`"
-      else
-        pure unit
+      else do
+        rmListener
+        mbSocket <- IM.socket response
+        for_ mbSocket \socket -> do
+          log "Destroying socket"
+          Stream.destroy (Socket.toDuplex socket)
+        log "testUpgrade - checkWebSocketUpgrade - Successfully upgraded to websocket."
+        done $ Right unit
+
     end (OM.toWriteable $ Client.toOutgoingMessage reqWSUpgrade)
+    pure nonCanceler
+