refactor(cli): clean up test runner channels (#22422)

Gets us closer to solving #20707.

Rewrites the `TestEventSender`:

- Allow for explicit creation of multiple streams. This will allow for
one-std{out,err}-per-worker
- All test events are received along with a worker ID, allowing for
eventual, proper parallel threading of test events.
 
In theory this should open up proper interleaving of test output,
however that is left for a future PR.

I had some plans for a better performing synchronization primitive, but
the inter-thread communication is tricky. This does, however, speed up
the processing of large numbers of tests 15-25% (possibly even more on
100,000+).

Before

```
ok | 1000 passed | 0 failed (32ms)
ok | 10000 passed | 0 failed (276ms)
```

After

```
ok | 1000 passed | 0 failed (25ms)
ok | 10000 passed | 0 failed (230ms)
```

2 files changed, 6 insertions, 8 deletions

diff --git a/cli/tools/repl/mod.rs b/cli/tools/repl/mod.rs
index e40c6362a..98519b60d 100644
--- a/cli/tools/repl/mod.rs
+++ b/cli/tools/repl/mod.rs
@@ -16,7 +16,6 @@ use deno_core::unsync::spawn_blocking;
 use deno_runtime::permissions::Permissions;
 use deno_runtime::permissions::PermissionsContainer;
 use rustyline::error::ReadlineError;
-use tokio::sync::mpsc::unbounded_channel;
 
 mod channel;
 mod editor;
@@ -32,8 +31,7 @@ pub use session::EvaluationOutput;
 pub use session::ReplSession;
 pub use session::REPL_INTERNALS_NAME;
 
-use super::test::TestEvent;
-use super::test::TestEventSender;
+use super::test::create_single_test_event_channel;
 
 struct Repl {
   session: ReplSession,
@@ -168,9 +166,8 @@ pub async fn run(flags: Flags, repl_flags: ReplFlags) -> Result<i32, AnyError> {
     .deno_dir()
     .ok()
     .and_then(|dir| dir.repl_history_file_path());
-  let (test_event_sender, test_event_receiver) =
-    unbounded_channel::<TestEvent>();
-  let test_event_sender = TestEventSender::new(test_event_sender);
+  let (worker, test_event_receiver) = create_single_test_event_channel();
+  let test_event_sender = worker.sender;
   let mut worker = worker_factory
     .create_custom_worker(
       main_module.clone(),
diff --git a/cli/tools/repl/session.rs b/cli/tools/repl/session.rs
index a52eb095f..65e27136f 100644
--- a/cli/tools/repl/session.rs
+++ b/cli/tools/repl/session.rs
@@ -14,6 +14,7 @@ use crate::tools::test::reporters::TestReporter;
 use crate::tools::test::run_tests_for_worker;
 use crate::tools::test::worker_has_tests;
 use crate::tools::test::TestEvent;
+use crate::tools::test::TestEventReceiver;
 use crate::tools::test::TestEventSender;
 
 use deno_ast::diagnostics::Diagnostic;
@@ -183,7 +184,7 @@ pub struct ReplSession {
   test_reporter_factory: Box<dyn Fn() -> Box<dyn TestReporter>>,
   test_event_sender: TestEventSender,
   /// This is only optional because it's temporarily taken when evaluating.
-  test_event_receiver: Option<tokio::sync::mpsc::UnboundedReceiver<TestEvent>>,
+  test_event_receiver: Option<TestEventReceiver>,
   jsx: ReplJsxState,
   experimental_decorators: bool,
 }
@@ -196,7 +197,7 @@ impl ReplSession {
     mut worker: MainWorker,
     main_module: ModuleSpecifier,
     test_event_sender: TestEventSender,
-    test_event_receiver: tokio::sync::mpsc::UnboundedReceiver<TestEvent>,
+    test_event_receiver: TestEventReceiver,
   ) -> Result<Self, AnyError> {
     let language_server = ReplLanguageServer::new_initialized().await?;
     let mut session = worker.create_inspector_session().await;