<commit_before>package client

import monifu.reactive.OverflowStrategy.DropNew
import monifu.reactive.{Observable, Subscriber}
import org.scalajs.dom
import shared.models.{Event, OverflowEvent, Signal}
import scala.concurrent.duration.FiniteDuration
import scala.scalajs.js.Dynamic.global

final class DataConsumer(interval: FiniteDuration, seed: Long, doBackPressure: Boolean)
  extends Observable[Event] {

  def onSubscribe(subscriber: Subscriber[Event]): Unit = {
    val host = dom.window.location.host

    val source = if (doBackPressure) {
      val url = s"ws://$host/back-pressured-stream?periodMillis=${interval.toMillis}&seed=$seed"
      BackPressuredWebSocketClient(url)
    }
    else {
      val url = s"ws://$host/simple-stream?periodMillis=${interval.toMillis}&seed=$seed"
      SimpleWebSocketClient(url, DropNew(1000))
    }

    source
      .collect { case IsEvent(e) => e }
      .onSubscribe(subscriber)
  }

  object IsEvent {
    def unapply(message: String) = {
      val json = global.JSON.parse(message)

      json.event.asInstanceOf[String] match {
        case "point" =>
          Some(Signal(
            value = json.value.asInstanceOf[Number].doubleValue(),
            timestamp = json.timestamp.asInstanceOf[Number].longValue()
          ))
        case "overflow" =>
          Some(OverflowEvent(
            dropped = json.dropped.asInstanceOf[Number].longValue(),<commit_msg>Fix web-socket link for secure connections<commit_after>package client

import monifu.reactive.OverflowStrategy.DropNew
import monifu.reactive.{Observable, Subscriber}
import org.scalajs.dom
import shared.models.{Event, OverflowEvent, Signal}
import scala.concurrent.duration.FiniteDuration
import scala.scalajs.js.Dynamic.global

final class DataConsumer(interval: FiniteDuration, seed: Long, doBackPressure: Boolean)
  extends Observable[Event] {

  def onSubscribe(subscriber: Subscriber[Event]): Unit = {
    val host = dom.window.location.host
    val protocol = if (dom.window.location.protocol == "https:") "wss:" else "ws:"

    val source = if (doBackPressure) {
      val url = s"$protocol//$host/back-pressured-stream?periodMillis=${interval.toMillis}&seed=$seed"
      BackPressuredWebSocketClient(url)
    }
    else {
      val url = s"$protocol//$host/simple-stream?periodMillis=${interval.toMillis}&seed=$seed"
      SimpleWebSocketClient(url, DropNew(1000))
    }

    source
      .collect { case IsEvent(e) => e }
      .onSubscribe(subscriber)
  }

  object IsEvent {
    def unapply(message: String) = {
      val json = global.JSON.parse(message)

      json.event.asInstanceOf[String] match {
        case "point" =>
          Some(Signal(
            value = json.value.asInstanceOf[Number].doubleValue(),
            timestamp = json.timestamp.asInstanceOf[Number].longValue()
          ))
        case "overflow" =>
          Some(OverflowEvent(
            dropped = json.dropped.asInstanceOf[Number].longValue(),
{'fraction_non_alphanumeric': 0.11840324763193505, 'fraction_numerical': 0.0027063599458728013, 'mean_word_length': 3.4667673716012084, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '3218087', 'n_tokens_mistral': 891, 'n_tokens_neox': 873, 'n_words': 164}