outwatch-test/src/main/scala/outwatchapp/util/reactive/ReconnectingWebSocket.scala

package outwatchapp.util.reactive

import scala.concurrent.Future

import io.circe.Decoder
import io.circe.Encoder
import io.circe.Printer
import io.circe.parser._
import io.circe.syntax._
import monix.bio.Task
import monix.execution.Ack
import monix.execution.Cancelable
import monix.execution.CancelablePromise
import monix.execution.cancelables.SingleAssignCancelable
import monix.reactive.Observable
import monix.reactive.Observer
import monix.reactive.OverflowStrategy
import typings.reconnectingWebsocket.eventsMod.CloseEvent
import typings.reconnectingWebsocket.eventsMod.ErrorEvent
import typings.reconnectingWebsocket.eventsMod.Event
import typings.reconnectingWebsocket.mod.{default => RW}

import scalajs.js
import typings.std.MessageEvent
import typings.reconnectingWebsocket.mod.Options
import outwatchapp.util.reactive.Exceptions.DecodeException
import outwatchapp.util.reactive.Exceptions.WrongTypeException
import monix.reactive.observers.Subscriber
import monix.reactive.observers.BufferedSubscriber
import outwatchapp.util.reactive.Exceptions.UseAfterClose
import monix.bio.IO

class ReconnectingWebSocketImpl[T: Encoder: Decoder](
    ws: RW,
    overflowStrategy: OverflowStrategy.Synchronous[T]
) {
  val printer = Printer.noSpaces

  /** @throws ReactiveException
    */
  val source: Task[Observable[T]] =
    Task.deferAction(implicit s =>
      for {
        obs <- Task(
          Observable
            .create[T](overflowStrategy) { sub =>
              val c = SingleAssignCancelable()
              val onmessage: js.Function1[MessageEvent[_], Unit] =
                _.data match {
                  case s: String =>
                    decode[T](s)
                      .map { res =>
                        if (sub.onNext(res) == Ack.Stop) c.cancel()
                        res
                      }
                      .left
                      .foreach(err =>
                        sub.onError(
                          DecodeException(
                            s"Failed to decode $s. Error was $err"
                          )
                        )
                      )
                  case other =>
                    sub.onError(
                      WrongTypeException(s"Received wrong type: $other")
                    )
                }
              val emptyMsgFn: js.Function1[MessageEvent[_], Unit] =
                _ => println("message fn not initialized")
              ws.onmessage = onmessage

              val onclose: js.Function1[CloseEvent, Unit] = a => {
                println(
                  s"Websocket closing -  ${a.code} ${a.wasClean} ${a.reason}"
                )
                sub.onComplete()
              }
              ws.onclose = onclose

              val onerror: js.Function1[ErrorEvent, Unit] = (e: Event) =>
                sub.onError(new Exception(s"Error in WebSocket: $e"))

              ws.onerror = onerror

              c := Cancelable { () => ws.onmessage = emptyMsgFn }
            }
            .publish
            .refCount
        )
        _ <- Task(obs.subscribe(Observer.empty))
      } yield obs
    )

  val sink: Task[Observer[T]] =
    Task.deferAction(implicit s =>
      Task(
        BufferedSubscriber(
          new Subscriber[T] {
            import monix.execution.FutureUtils
            import scala.concurrent.duration._
            override def scheduler = s
            override def onNext(elem: T): Future[Ack] = {
              val msg = printer.print(elem.asJson)
              if (ws.readyState == 2 || ws.readyState == 3)
                Ack.Stop
              else {
                FutureUtils
                  .delayedResult(2.second)(ws.send(msg))
                  .flatMap(_ => Ack.Continue)
              }
            }
            override def onError(ex: Throwable): Unit = s.reportFailure(ex)
            override def onComplete(): Unit = println("CLOSING WEBSOCKET 2 ")
          },
          OverflowStrategy.Default
        )
      )
    )

}
object ReconnectingWebSocket {
  sealed trait Error
  final case object Error extends Error

  /** @throws ReactiveException
    */
  final case class Source[T](value: Observable[T])

  /** A buffered Observer
    *
    * @param value
    */
  final case class Sink[T](value: Observer[T]) {
    def send(t: T) =
      IO.deferFuture(value.onNext(t)).flatMap {
        case Ack.Continue => Task.unit
        case Ack.Stop     => IO.raiseError(UseAfterClose("Websocket was closed"))
      }
  }

  private val defaultOptions = Options()
    .setMinReconnectionDelay(5000)
    .setMaxRetries(2)
    .setMaxEnqueuedMessages(50)

  def onopen[A](op: => A): js.Function1[Event, Unit] = _ => op

  def apply[T <: Product: Encoder: Decoder](
      path: String,
      options: Options = defaultOptions,
      overflowStrategy: OverflowStrategy.Synchronous[T] =
        OverflowStrategy.DropOld(50)
  ) = for {
    websocket <- Task(new RW(path, js.undefined, options))
    p = CancelablePromise[Unit]()
    _ <- Task(websocket.onopen = onopen(p.success(())))
    _ <- Task.deferFuture(p.future)
    _ <- Task(websocket.onopen = onopen(()))
    impl = new ReconnectingWebSocketImpl[T](websocket, overflowStrategy)
    source <- impl.source
    sink <- impl.sink
    // _ <- Task.deferAction(implicit s => Task(source.subscribe(sink)))
  } yield Source(source) -> Sink(sink)
}