Slick 3 DatabasePublisher to Rx Observable

slick-rx.scala

import rx.lang.scala.{Subject, Observable}
import slick.backend.DatabasePublisher

import scala.concurrent.ExecutionContext
import scala.util.{Failure, Success}

trait StreamExtensions {
  implicit def extendDatabasePublisher[T](publisher: DatabasePublisher[T]): StreamExtensions.DatabasePublisherExtensions[T] =
    StreamExtensions.DatabasePublisherExtensions[T](publisher)
}

object StreamExtensions {
  implicit class DatabasePublisherExtensions[T](val publisher: DatabasePublisher[T]) extends AnyVal {
    def toObservable(implicit ec: ExecutionContext): Observable[T] = {
      val subject = Subject[T]()
      val future = publisher foreach subject.onNext
      future onComplete {
        case Success(_) => subject.onCompleted()
        case Failure(error) => subject.onError(error)
      }
      subject
    }
  }
}