<commit_before>package client

import monix.execution.{Ack, Cancelable, Scheduler}
import monix.reactive.observers.Subscriber
import monix.reactive.{Observable, Observer, OverflowStrategy, Pipe}
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.control.NonFatal


final class SimpleWebSocketClient private(url: String, os: OverflowStrategy.Synchronous[String])
  extends Observable[String] {
  self =>

  private def createChannel()(implicit s: Scheduler): Observable[String] = {

    var webSocket: WebSocket = null
    try {
      Utils.log(s"Connecting to $url")
      webSocket = new WebSocket(url)
    } catch {
      case NonFatal(ex) =>
        Observable.raiseError(ex)
    }

    def closeConnection(): Unit = {
      if (webSocket != null && webSocket.readyState <= 1)
        try webSocket.close() catch {
          case _: Throwable => ()
        }
    }

    try {
      val (in, out) = Pipe.publish[String].concurrent(os)
      webSocket.onopen = (event: Event) => ()

      webSocket.onerror = (event: ErrorEvent) => {
        in.onError(BackPressuredWebSocketClient.Exception(event.message))
      }

      webSocket.onclose = (event: CloseEvent) => {
        in.onComplete()
      }

      webSocket.onmessage = (event: MessageEvent) => {
        in.onNext(event.data.asInstanceOf[String])
      }

      out.doOnCancel(closeConnection())
    } catch {
      case NonFatal(ex) =>
        Observable.raiseError(ex)
    }
  }

  override def unsafeSubscribeFn(subscriber: Subscriber[String]): Cancelable = {
    import subscriber.scheduler

    val subscription: Cancelable =
      createChannel().unsafeSubscribeFn(new Observer[String] {
        def onNext(elem: String): Future[Ack] =
          subscriber.onNext(elem)

        def onError(ex: Throwable): Unit = {
          subscription.cancel()
          scheduler.reportFailure(ex)
          // retry connection in a couple of secs
          self
            .delaySubscription(3.seconds)
            .unsafeSubscribeFn(subscriber)
        }<commit_msg>using doOnDownstreamStop instead of doOnCancel<commit_after>package client

import monix.execution.{Ack, Cancelable, Scheduler}
import monix.reactive.observers.Subscriber
import monix.reactive.{Observable, Observer, OverflowStrategy, Pipe}
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.control.NonFatal


final class SimpleWebSocketClient private(url: String, os: OverflowStrategy.Synchronous[String])
  extends Observable[String] {
  self =>

  private def createChannel()(implicit s: Scheduler): Observable[String] = {

    var webSocket: WebSocket = null
    try {
      Utils.log(s"Connecting to $url")
      webSocket = new WebSocket(url)
    } catch {
      case NonFatal(ex) =>
        Observable.raiseError(ex)
    }

    def closeConnection(): Unit = {
      if (webSocket != null && webSocket.readyState <= 1)
        try webSocket.close() catch {
          case _: Throwable => ()
        }
    }

    try {
      val (in, out) = Pipe.publish[String].concurrent(os)
      webSocket.onopen = (event: Event) => ()

      webSocket.onerror = (event: ErrorEvent) => {
        in.onError(BackPressuredWebSocketClient.Exception(event.message))
      }

      webSocket.onclose = (event: CloseEvent) => {
        in.onComplete()
      }

      webSocket.onmessage = (event: MessageEvent) => {
        in.onNext(event.data.asInstanceOf[String])
      }

      out.doOnDownstreamStop(closeConnection())
    } catch {
      case NonFatal(ex) =>
        Observable.raiseError(ex)
    }
  }

  override def unsafeSubscribeFn(subscriber: Subscriber[String]): Cancelable = {
    import subscriber.scheduler

    val subscription: Cancelable =
      createChannel().unsafeSubscribeFn(new Observer[String] {
        def onNext(elem: String): Future[Ack] =
          subscriber.onNext(elem)

        def onError(ex: Throwable): Unit = {
          subscription.cancel()
          scheduler.reportFailure(ex)
          // retry connection in a couple of secs
          self
            .delaySubscription(3.seconds)
            .unsafeSubscribeFn(subscriber)
        }
{'fraction_non_alphanumeric': 0.10940718699931334, 'fraction_numerical': 0.0009155413138017853, 'mean_word_length': 2.9192825112107625, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 19, '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': '370081', 'n_tokens_mistral': 1323, 'n_tokens_neox': 1275, 'n_words': 260}