//设置package包名称以及导入依赖的类
package walfie.gbf.raidfinder.util

import monix.execution.Ack
import monix.reactive.Observer
import scala.concurrent.Future

class TestObserver[T] extends Observer[T] {
  var received = Vector.empty[T]
  var errorsReceived = Vector.empty[Throwable]
  var isComplete: Boolean = false

  def onNext(elem: T): Future[Ack] = {
    received = received :+ elem
    Ack.Continue
  }

  def onError(e: Throwable): Unit = {
    errorsReceived = errorsReceived :+ e
  }

  def onComplete(): Unit = {
    isComplete = true
  }
} 

//设置package包名称以及导入依赖的类
package my.samples.observers

import akka.actor.ActorRef
import monix.execution.Ack.Continue
import monix.execution.{ Ack, Scheduler }
import monix.reactive.observers.Subscriber
import my.samples.models.MyMessages
import my.samples.models.MyMessages.Tick
import org.slf4j.LoggerFactory

import scala.concurrent.Future

class MyObserver(actorRef: ActorRef, sourceName: String)(implicit s: Scheduler) extends Subscriber[Long] {

  private[this] def logger = LoggerFactory.getLogger(this.getClass)

  override implicit def scheduler: Scheduler = s

  override def onError(ex: Throwable): Unit =
    logger.error(s"error happened when processing the stream: error message << ${ex.getMessage} >>")

  override def onComplete(): Unit =
    logger.info("stream completed")

  override def onNext(elem: Long): Future[Ack] = {
    logger.info(s"message received from source $sourceName --> $elem")
    actorRef ! Tick(sourceName, elem)
    Continue
  }
}
object MyObserver {
  def apply(actorRef: ActorRef, sourceName: String)(implicit s: Scheduler) = {
    new MyObserver(actorRef, sourceName)(s)
  }
} 

//设置package包名称以及导入依赖的类
package com.hypertino.hyperbus.util

import com.hypertino.hyperbus.transport.api.matchers.RequestMatcher
import monix.eval.Task
import monix.execution.Ack.Stop
import monix.execution.{Ack, Cancelable, Scheduler}
import monix.reactive.{Observable, Observer}
import monix.reactive.observers.Subscriber
import monix.reactive.subjects.{ConcurrentSubject, Subject}

import scala.util.Success

abstract class SubjectSubscription[T](implicit val scheduler: Scheduler) extends FuzzyMatcher {
  type eventType = T

  // FyzzyIndex properties
  def requestMatcher: RequestMatcher
  override def indexProperties: Seq[FuzzyIndexItemMetaInfo] = requestMatcher.indexProperties
  override def matches(other: Any): Boolean = requestMatcher.matches(other)

  // Subject properties
  protected val subject: Subject[eventType, eventType]
  def cancel(): Unit = {
    remove()
    subject.onComplete()
  }

  def publish(t: eventType): Task[Ack] = {
    Task.fromFuture(subject.onNext(t).andThen {
      case Success(Stop) ? remove()
    })
  }

  private def cancel_1() = cancel()

  val observable: Observable[eventType] = new Observable[eventType] {
    override def unsafeSubscribeFn(subscriber: Subscriber[eventType]): Cancelable = {
      val original: Cancelable = subject.unsafeSubscribeFn(subscriber)
      add()
      new Cancelable {
        override def cancel(): Unit = {
          cancel_1()
          original.cancel()
        }
      }
    }
  }

  protected def remove(): Unit
  protected def add(): Unit
} 

//设置package包名称以及导入依赖的类
package flumina.monix

import flumina.{KafkaResult, TopicPartitionValue}
import monix.eval.Task
import monix.execution.Ack
import scodec.Err

import scala.concurrent.Future
import scala.concurrent.duration.FiniteDuration

final class TopicErrorException(errors: List[TopicPartitionValue[KafkaResult]]) extends Exception

sealed trait ConsumptionStrategy

object ConsumptionStrategy {
  final case class Tail(delay: FiniteDuration) extends ConsumptionStrategy
  case object TerminateEndOfStream             extends ConsumptionStrategy
}

trait CodecErrorHandler {
  def handle(err: Err): Future[Ack]
}

object CodecErrorHandler {
  val stop: CodecErrorHandler = new CodecErrorHandler {
    override def handle(err: Err): Future[Ack] = Ack.Stop
  }
} 

//设置package包名称以及导入依赖的类
package de.m7w3.signal.events

import de.m7w3.signal.Logging
import monix.execution.Ack
import monix.execution.Ack.Continue
import monix.reactive.Observer

import scala.concurrent.Future

trait EventListener extends Observer[SignalDesktopEvent] {
  def handle: PartialFunction[SignalDesktopEvent, Unit]

  override def onNext(elem: SignalDesktopEvent): Future[Ack] = {
    if (handle.isDefinedAt(elem)) {
      handle.apply(elem)
    }
    Continue
  }
}


abstract class SimpleEventListener extends EventListener with Logging {

  override def onError(ex: Throwable): Unit = {
    logger.error(s"Error in event listener ${getClass.getSimpleName}", ex)
  }

  override def onComplete(): Unit = {
    logger.info(s"${getClass.getSimpleName} completed")
  }
}

object EventListener {
  def simple(onEvent: PartialFunction[SignalDesktopEvent, Unit]): EventListener = {
    new SimpleEventListener() {
      override def handle: PartialFunction[SignalDesktopEvent, Unit] = onEvent
    }
  }
} 

//设置package包名称以及导入依赖的类
package de.m7w3.signal.events

import de.m7w3.signal.Logging
import monix.execution.Ack.Stop
import monix.execution.Scheduler.Implicits.global
import monix.execution.atomic.{Atomic, AtomicBoolean}
import monix.execution.{Ack, Cancelable}
import monix.reactive.OverflowStrategy.DropOld
import monix.reactive.subjects.ConcurrentSubject

import scala.concurrent.Future


trait EventPublisher {
  
  def publishEvent(event: SignalDesktopEvent): Future[Ack]
}

trait EventDispatcher {
  def register(listener: EventListener): Cancelable
}

class SignalDesktopEventDispatcher extends EventPublisher with EventDispatcher with Logging {

  private val alive: AtomicBoolean = Atomic(true)
  private val bufferSize: Int = 1024
  private val overflowStrategy = DropOld(bufferSize)
  // ensures concurrency,safety
  // and backpressure by maintaining a buffer
  // in this subject that sits before any subscriber
  // but has the drawback that we might execute stuff asynchronously
  private val subject: ConcurrentSubject[SignalDesktopEvent, SignalDesktopEvent] =
    ConcurrentSubject.publish[SignalDesktopEvent](overflowStrategy)

  override def publishEvent(event: SignalDesktopEvent): Future[Ack] = {
    if (alive.get) {
       subject.onNext(event)
    } else {
      logger.warn(s"closed. ignoring event $event")
      Stop
    }
  }

  override def register(listener: EventListener): Cancelable = {
    subject.subscribe(listener)
  }

  def close(): Unit = {
    alive.set(false)
    subject.onComplete()
  }
}