<commit_before>package client

import client.WebSocketClient.SocketException
import monix.execution.Ack.{Continue, Stop}
import monix.execution.cancelables.SingleAssignmentCancelable
import monix.execution.{Ack, Cancelable}
import monix.reactive.observers.Subscriber
import monix.reactive.{Observable, OverflowStrategy}
import org.scalajs.dom
import org.scalajs.dom.raw.MessageEvent
import org.scalajs.dom.{CloseEvent, ErrorEvent, Event, WebSocket}

import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.Try
import scala.util.control.NonFatal

class WebSocketClient private(url: String, messages: Either[String, () => Observable[String]]) extends Observable[String] {

  def unsafeSubscribeFn(subscriber: Subscriber[String]): Cancelable = {
    val inbound: Observable[String] =
      Observable.create[String](OverflowStrategy.DropOld(100)) { downstream =>
        val cancelOutbound = SingleAssignmentCancelable()

        try {
          val webSocket = new WebSocket(url)

          def closeConnection(): Unit = {
            cancelOutbound.cancel()
            if (webSocket.readyState <= 1) Try(webSocket.close())
          }

          webSocket.onopen = (event: Event) => messages match {
            case Left(message) => webSocket.send(message)
            case Right(ms) =>
              implicit val s = subscriber.scheduler
              cancelOutbound := ms().subscribe({ m =>
                webSocket.send(m)
                Continue
              }, _ => closeConnection(), closeConnection)
          }
          webSocket.onerror = (event: ErrorEvent) => {
            cancelOutbound.cancel()
            downstream.onError(SocketException(event.message))
          }
          webSocket.onclose = (event: CloseEvent) => {
            cancelOutbound.cancel()
            downstream.onComplete()
          }
          webSocket.onmessage = (event: MessageEvent) => {
            val ack = downstream.onNext(event.data.asInstanceOf[String])
            if (ack == Stop) closeConnection()
          }

          Cancelable(closeConnection)
        } catch {
          case NonFatal(ex) =>
            downstream.onError(ex)
            Cancelable.empty
        }
      }<commit_msg>add delay for SolutionTemplate calc<commit_after>package client

import client.WebSocketClient.SocketException
import monix.execution.Ack.{Continue, Stop}
import monix.execution.cancelables.SingleAssignmentCancelable
import monix.execution.{Ack, Cancelable, FutureUtils}
import monix.reactive.observers.Subscriber
import monix.reactive.{Observable, OverflowStrategy}
import org.scalajs.dom
import org.scalajs.dom.raw.MessageEvent
import org.scalajs.dom.{CloseEvent, ErrorEvent, Event, WebSocket}

import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.Try
import scala.util.control.NonFatal

class WebSocketClient private(url: String, messages: Either[String, () => Observable[String]]) extends Observable[String] {

  def unsafeSubscribeFn(subscriber: Subscriber[String]): Cancelable = {
    val inbound: Observable[String] =
      Observable.create[String](OverflowStrategy.DropOld(100)) { downstream =>
        val cancelOutbound = SingleAssignmentCancelable()

        try {
          val webSocket = new WebSocket(url)

          def closeConnection(): Unit = {
            cancelOutbound.cancel()
            if (webSocket.readyState <= 1) Try(webSocket.close())
          }

          webSocket.onopen = (event: Event) => messages match {
            case Left(message) => webSocket.send(message)
            case Right(ms) =>
              implicit val s = subscriber.scheduler
              cancelOutbound := ms().subscribe({ m =>
                webSocket.send(m)
                FutureUtils.delayedResult(1.second)(Continue)
              }, _ => closeConnection(), closeConnection)
          }
          webSocket.onerror = (event: ErrorEvent) => {
            cancelOutbound.cancel()
            downstream.onError(SocketException(event.message))
          }
          webSocket.onclose = (event: CloseEvent) => {
            cancelOutbound.cancel()
            downstream.onComplete()
          }
          webSocket.onmessage = (event: MessageEvent) => {
            val ack = downstream.onNext(event.data.asInstanceOf[String])
            if (ack == Stop) closeConnection()
          }

          Cancelable(closeConnection)
        } catch {
          case NonFatal(ex) =>
            downstream.onError(ex)
            Cancelable.empty
        }
      }
{'fraction_non_alphanumeric': 0.10269434628975264, 'fraction_numerical': 0.001987632508833922, 'mean_word_length': 2.6116427432216907, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2884730', 'n_tokens_mistral': 1235, 'n_tokens_neox': 1193, 'n_words': 235}