36 of 39 passed

Author: gmkumar2005

src/main/scala/jsenv/playwright/CEEnv.scala

@@ -1,7 +1,7 @@
 package jsenv.playwright
 
 import org.scalajs.jsenv._
-import scribe.format.{FormatterInterpolator, date, level, mdc, messages, methodName, threadName}
+import scribe.format.{FormatterInterpolator, dateFull, level, mdc, messages, methodName, threadName}
 
 import scala.util.control.NonFatal
 
@@ -11,6 +11,8 @@ class CEEnv(
     showLogs: Boolean = false,
     pwConfig: PWEnv.Config
 ) extends JSEnv {
+//  private lazy val validator = ExternalJSRun.supports(RunConfig.Validator())
+
   private lazy val validator = {
     RunConfig
       .Validator()
@@ -22,26 +24,44 @@ class CEEnv(
 
   override def start(input: Seq[Input], runConfig: RunConfig): JSRun =
     try {
+      scribe.info(
+        s"Calling validator CEEnv.start with input $input and runConfig $runConfig"
+      )
       validator.validate(runConfig)
       new CERun(pwConfig, runConfig, input)
     } catch {
-      case NonFatal(t) =>
-        JSRun.failed(t)
+      case e: Exception =>
+        scribe.error(s"IllegalArgumentException $e")
+        throw e
+      case r : RuntimeException =>
+        scribe.error(s"RuntimeException $r")
+        throw r
+
+//      case NonFatal(t) =>
+//        scribe.error(s"CEEnv.start failed with $t")
+//        JSRun.failed(t)
     }
 
   override def startWithCom(
       input: Seq[Input],
       runConfig: RunConfig,
       onMessage: String => Unit
-  ): JSComRun = new CEComRun(pwConfig, runConfig, input, onMessage)
+  ): JSComRun = try {
+    validator.validate(runConfig)
+    new CEComRun(pwConfig, runConfig, input, onMessage)
+  } catch {
+    case NonFatal(t) =>
+      JSComRun.failed(t)
+  }
 
   private def setupLogger(showLogs: Boolean): Unit = {
     val formatter =
-      formatter"$date [$threadName] $level $methodName - $messages$mdc"
+      formatter"$dateFull [$threadName] $level $methodName - $messages$mdc"
+//    val formatter"${date("yyyy-MM-dd HH:mm:ss.SSS")} [$threadName] $level $methodName - $messages$mdc"
     scribe.Logger.root
       .clearHandlers()
       .withHandler(
-        formatter = formatter,
+        formatter = formatter
       )
       .replace()
     if (showLogs) {

src/main/scala/jsenv/playwright/CERun.scala

@@ -20,7 +20,13 @@ class CERun(
 
   implicit val ec: scala.concurrent.ExecutionContext =
     scala.concurrent.ExecutionContext.global
-
+  private lazy val validator = {
+    RunConfig
+      .Validator()
+      .supportsInheritIO()
+      .supportsOnOutputStream()
+  }
+  validator.validate(runConfig)
 //  private[this] implicit val ec =
 //    ExecutionContext.fromExecutor(Executors.newSingleThreadExecutor())
 
@@ -30,6 +36,7 @@ class CERun(
   val stopSignal: Deferred[IO, Boolean] = Deferred.unsafe[IO, Boolean]
   stopSignal.complete(false).unsafeRunSync()
   protected val sendQueue = new ConcurrentLinkedQueue[String]
+  protected def receivedMessage(msg: String): Unit = ()
   System.setProperty("playwright.driver.impl", "jsenv.DriverJar")
 
   /** A [[scala.concurrent.Future Future]] that completes if the run completes.
@@ -59,31 +66,35 @@ class CERun(
   // If want to close then close driver, streams, materializer
   // After future is completed close driver, streams, materializer
 
-  def jsRunPrg(isComEnabled:Boolean): Resource[IO, Unit] = for {
-    _ <- Resource.pure(scribe.info(s"Begin Main JSRun!"))
+  def jsRunPrg(isComEnabled: Boolean): Resource[IO, Unit] = for {
+    _ <- Resource.pure(
+      scribe.info(s"Begin Main JSRun! with isComEnabled $isComEnabled")
+    )
     pageInstance <- createPage(headlessConfig)
     _ <- preparePageForJsRun(
       pageInstance,
       materializer(pwConfig),
       input,
       isComEnabled
     )
-    initialStatus <- isConnectionUp(pageInstance, intf)
+    connectionReady <- isConnectionUp(pageInstance, intf)
     _ <-
-      if (!initialStatus) Resource.pure[IO, Unit](IO.sleep(100.milliseconds))
+      if (!connectionReady) Resource.pure[IO, Unit](IO.sleep(100.milliseconds))
       else Resource.pure[IO, Unit](IO.unit)
     _ <- isConnectionUp(pageInstance, intf)
-    jsResponse <- fetchMessages(pageInstance, intf)
-    _ <- Resource.pure(scribe.info(s"jsResponse is $jsResponse"))
-    _ <- streamWriter(jsResponse, runConfig)
-    _ <- Resource.pure(
-      scribe.info(
-        s"StopSignal in jsRunPrg is ${stopSignal.get.unsafeRunSync()}"
-      )
+    jsResponse <- fetchMessagesResource(pageInstance, intf)
+    _ <- Resource.pure(scribe.debug(s"jsResponse is $jsResponse"))
+    out <- ResourcesFactory.outputStream(runConfig)
+    _ <- Resource.pure[IO, Unit](
+      streamWriter(jsResponse, out, Some(receivedMessage))
     )
-    _ <- sendAllResource(sendQueue, pageInstance, intf)
-    _ <- fetchAndProcess(stopSignal, pageInstance, intf, runConfig)
 
+    //    while (!wantClose) {
+    //    sendAll()
+    //    fetchAndProcess()
+    //    Thread.sleep(100)
+    //  }
+    _ <- ProcessUntilStop(wantToClose, pageInstance, intf, sendQueue, out, receivedMessage, isComEnabled)
   } yield {
     handleErrors(jsResponse)
   }
@@ -102,7 +113,8 @@ class CERun(
     }
   }
 
-  val future: Future[Unit] = jsRunPrg(enableCom).use(_ => IO.unit).unsafeToFuture()
+  lazy val future: Future[Unit] =
+    jsRunPrg(enableCom).use(_ => IO.unit).unsafeToFuture()
 
   /** Stops the run and releases all the resources.
     *
@@ -120,7 +132,7 @@ class CERun(
 
   override def close(): Unit = {
     wantToClose.set(true)
-    stopSignal.complete(true).unsafeRunSync()
+    stopSignal.complete(true)
 
     scribe.info(s"WantToClose in close is ${wantToClose.get()}")
   }
@@ -140,7 +152,9 @@ class CEComRun(
     scribe.info("Send a message to sendQueue")
     sendQueue.offer(msg)
   }
-  override val future: Future[Unit] = jsRunPrg(enableCom).use(_ => IO.unit).unsafeToFuture()
+  override protected def receivedMessage(msg: String): Unit = onMessage(msg)
+//  override lazy val future: Future[Unit] =
+//    jsRunPrg(enableCom).use(_ => IO.unit).unsafeToFuture()
 }
 
 private class WindowOnErrorException(errs: List[String])

src/main/scala/jsenv/playwright/PwRun.scala

@@ -85,12 +85,15 @@ private final class PwComRun(
     with JSComRun {
   private[this] val sendQueue = new ConcurrentLinkedQueue[String]
 
-  def send(msg: String): Unit = sendQueue.offer(msg)
+  def send(msg: String): Unit = {
+    scribe.info(s"Calling send with message $msg")
+    sendQueue.offer(msg)}
 
   override protected def receivedMessage(msg: String): Unit = onMessage(msg)
 
   @tailrec
   override protected def sendAll(): Unit = {
+    scribe.info(s"Calling sendAll")
     val msg = sendQueue.poll()
     if (msg != null) {
       scribe.info(s"Sending message $msg")

src/main/scala/jsenv/playwright/ResourcesFactory.scala

@@ -7,7 +7,7 @@ import org.scalajs.jsenv.{Input, RunConfig}
 
 import java.util
 import java.util.concurrent.ConcurrentLinkedQueue
-import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
 import java.util.function.Consumer
 import scala.annotation.tailrec
 import scala.concurrent.duration.DurationInt
@@ -102,21 +102,26 @@ object ResourcesFactory {
     } yield ()
 
   val fetchCounter = new AtomicInteger(0)
+
   def fetchMessages(
       pageInstance: Page,
       intf: String
+  ): java.util.Map[String, java.util.List[String]] = {
+    val data =
+      pageInstance
+        .evaluate(s"$intf.fetch();")
+        .asInstanceOf[java.util.Map[String, java.util.List[String]]]
+    data
+  }
+  def fetchMessagesResource(
+      pageInstance: Page,
+      intf: String
   ): Resource[IO, util.Map[String, util.List[String]]] = {
-    Resource.make {
-      IO {
-        scribe.info(
-          s"Page instance is ${pageInstance.hashCode()} fetchCounter is ${fetchCounter.incrementAndGet()}"
-        )
-        val data = pageInstance
-          .evaluate(s"$intf.fetch();")
-          .asInstanceOf[java.util.Map[String, java.util.List[String]]]
-        data
-      }
-    }(_ => IO.unit)
+    Resource.pure[IO, util.Map[String, util.List[String]]] {
+      scribe.info(
+          s"Page instance is ${pageInstance.hashCode()} fetchCounter is ${fetchCounter.incrementAndGet()}")
+        fetchMessages(pageInstance, intf)
+    }
   }
 
   def sendAllResource(
@@ -125,31 +130,60 @@ object ResourcesFactory {
       intf: String
   ): Resource[IO, Unit] = {
     Resource.make {
-      scribe.info(s"Sending all messages sendQueue size is ${sendQueue.size()} ")
+      scribe.info(
+        s"Sending all messages sendQueue size is ${sendQueue.size()} "
+      )
       sendAll(sendQueue, pageInstance, intf)
       IO.unit
     }(_ => IO.unit)
   }
 
+  def ProcessUntilStop(
+      stopSignal: AtomicBoolean,
+      pageInstance: Page,
+      intf: String,
+      sendQueue: ConcurrentLinkedQueue[String],
+      outStream: OutputStreams.Streams,
+      receivedMessage: String => Unit,
+      isComEnabled: Boolean
+  ): Resource[IO, Unit] = {
+    Resource.make {
+      IO.unit
+    } { _ =>
+      while (!stopSignal.get() && isComEnabled) {
+        scribe.info(
+          s"Calling repeatSendUntilStopSignal with stopSignal = ${stopSignal.get()}"
+        )
+        IO.sleep(100.milliseconds)
+        sendAll(sendQueue, pageInstance, intf)
+        val jsResponse = fetchMessages(pageInstance, intf)
+        streamWriter(jsResponse, outStream, Some(receivedMessage))
+      }
+      IO.unit
+    }
+  }
+
+
   def fetchAndProcess(
       stopSignal: Deferred[IO, Boolean],
       pageInstance: Page,
       intf: String,
-      runConfig: RunConfig
+//      runConfig: RunConfig,
+      outStream: OutputStreams.Streams
   ): Resource[IO, Unit] = {
     Resource.make {
-      stopSignal.get.attempt.flatMap {
-        case Left(_) =>
+      stopSignal.get.flatMap {
+        case true =>
           scribe.info("Stopping the program")
           IO.unit
-        case Right(_) => {
+        case false => {
           scribe.info("Program is running")
           for {
             _ <- Resource.pure(
               scribe.info(s"Page instance is ${pageInstance.hashCode()}")
             )
-            jsResponse <- fetchMessages(pageInstance, intf)
-            _ <- streamWriter(jsResponse, runConfig)
+            jsResponse <- fetchMessagesResource(pageInstance, intf)
+            _ <- streamWriterResource(jsResponse, outStream)
           } yield ()
           IO.sleep(100.milliseconds)
         }
@@ -161,14 +195,15 @@ object ResourcesFactory {
       pageInstance: Page,
       intf: String
   ): Resource[IO, Boolean] = {
-    Resource.make {
-      IO {
-        scribe.info(s"Page instance is ${pageInstance.hashCode()}")
-        pageInstance.evaluate(s"!!$intf;").asInstanceOf[Boolean]
-      }
-    }(_ => IO.unit)
+    Resource.pure[IO, Boolean] {
+      scribe.info(s"Page instance is ${pageInstance.hashCode()}")
+      pageInstance.evaluate(s"!!$intf;").asInstanceOf[Boolean]
+    }
+
   }
 
+
+
   def materializer(pwConfig: Config): Resource[IO, FileMaterializer] =
     Resource.make {
       IO.blocking(FileMaterializer(pwConfig.materialization)) // build
@@ -185,7 +220,7 @@ object ResourcesFactory {
   /*
    * Creates resource for outputStream
    */
-  private def outputStream(
+  def outputStream(
       runConfig: RunConfig
   ): Resource[IO, OutputStreams.Streams] =
     Resource.make {
@@ -202,24 +237,37 @@ object ResourcesFactory {
 
   def streamWriter(
       jsResponse: util.Map[String, util.List[String]],
-      runConfig: RunConfig,
+      outStream: OutputStreams.Streams,
       onMessage: Option[String => Unit] = None
-  ): Resource[IO, Unit] = {
+  ): Unit = {
     val data = jsResponse.get("consoleLog")
     val consoleError = jsResponse.get("consoleError")
     val error = jsResponse.get("errors")
     onMessage match {
       case Some(f) =>
         val msgs = jsResponse.get("msgs")
+        //        data.get("msgs").forEach(consumer(receivedMessage _))
+        //        msgs.forEach(receivedMessage(f))
         msgs.forEach(consumer(f))
-      case None => ()
+      case None => scribe.info("No onMessage function")
     }
-    for {
-      out <- ResourcesFactory.outputStream(runConfig)
-      _ <- Resource.pure(data.forEach(out.out.println _))
-      _ <- Resource.pure(error.forEach(out.out.println _))
-      _ <- Resource.pure(consoleError.forEach(out.out.println _))
-    } yield ()
+    data.forEach(outStream.out.println _)
+    error.forEach(outStream.out.println _)
+    consoleError.forEach(outStream.out.println _)
+//    val errs = data.get("errors")
+    if (!error.isEmpty) {
+      // Convoluted way of writing errs.toList without JavaConverters.
+      val errList = error.toArray(Array[String]()).toList
+      throw new WindowOnErrorException(errList)
+    }
+  }
+  def streamWriterResource(
+      jsResponse: util.Map[String, util.List[String]],
+      outStream: OutputStreams.Streams,
+      onMessage: Option[String => Unit] = None
+  ): Resource[IO, Unit] = {
+    streamWriter(jsResponse, outStream, onMessage)
+    Resource.pure[IO, Unit](IO.unit)
   }
 
   @tailrec
@@ -229,8 +277,8 @@ object ResourcesFactory {
       intf: String
   ): Unit = {
     val msg = sendQueue.poll()
-    scribe.info(s"Sending message $msg")
     if (msg != null) {
+      scribe.debug(s"Sending message $msg")
       val script = s"$intf.send(arguments[0]);"
       val wrapper = s"function(arg) { $script }"
       pageInstance.evaluate(s"$wrapper", msg)