Add entropysource to newInternalHttpClient so websockets work, and fix test to actually exercise websocket path of newInternalHttpClient (#5403)

Co-authored-by: Matt Alonso <[email protected]>

Author: tewaro

src/workerd/api/sockets.c++

@@ -531,9 +531,12 @@ kj::Own<kj::AsyncIoStream> Socket::takeConnectionStream(jsg::Lock& js) {
 // Implementation of the custom factory for creating WorkerInterface instances from a socket
 class StreamOutgoingFactory final: public Fetcher::OutgoingFactory, public kj::Refcounted {
  public:
-  StreamOutgoingFactory(kj::Own<kj::AsyncIoStream> stream, const kj::HttpHeaderTable& headerTable)
+  StreamOutgoingFactory(kj::Own<kj::AsyncIoStream> stream,
+      kj::EntropySource& entropySource,
+      const kj::HttpHeaderTable& headerTable)
       : stream(kj::mv(stream)),
-        httpClient(kj::newHttpClient(headerTable, *this->stream)) {}
+        httpClient(
+            kj::newHttpClient(headerTable, *this->stream, {.entropySource = entropySource})) {}
 
   kj::Own<WorkerInterface> newSingleUseClient(kj::Maybe<kj::String> cfStr) override;
 
@@ -621,7 +624,7 @@ jsg::Promise<jsg::Ref<Fetcher>> SocketsModule::internalNewHttpClient(
 
         // Create our custom factory that will create client instances from this socket
         kj::Own<Fetcher::OutgoingFactory> outgoingFactory = kj::refcounted<StreamOutgoingFactory>(
-            socket->takeConnectionStream(js), ioctx.getHeaderTable());
+            socket->takeConnectionStream(js), ioctx.getEntropySource(), ioctx.getHeaderTable());
 
         // Create a Fetcher that uses our custom factory
         auto fetcher = js.alloc<Fetcher>(

src/workerd/api/tests/http-socket-test.js

@@ -261,8 +261,24 @@ export const websockets = {
     const socket = connect(`localhost:${env.HTTP_SOCKET_SERVER_PORT}`);
     const httpClient = await internalNewHttpClient(socket);
 
-    // Create a WebSocket connection using the HTTP client for fetch
-    const ws = new WebSocket(`ws://localhost:${env.HTTP_SOCKET_SERVER_PORT}/`);
+    const res = await fetch(
+      `http://localhost:${env.HTTP_SOCKET_SERVER_PORT}/`,
+      {
+        headers: {
+          Connection: 'Upgrade',
+          Upgrade: 'websocket',
+        },
+      }
+    );
+
+    const ws = res.webSocket;
+
+    if (!ws) {
+      throw new Error('WebSocket upgrade failed');
+    }
+
+    ws.accept();
+    ws.send('Hello from test client');
 
     // Test the WebSocket connection
     await new Promise((resolve, reject) => {
@@ -271,10 +287,6 @@ export const websockets = {
         reject(new Error('WebSocket test timed out after 5 seconds'));
       }, 5000);
 
-      ws.addEventListener('open', () => {
-        ws.send('Hello from test client');
-      });
-
       ws.addEventListener('message', (event) => {
         // Verify we got the welcome message
         if (event.data === 'Welcome to WebSocket server') {