// SET UP CONNECTION AND POLLER
val host      = config.getString("db.host").get
val port      = config.getInt("db.port").get
val dbName    = config.getString("db.name").get
val user      = config.getString("slick.dbs.default.db.user").get
val password  = config.getString("slick.dbs.default.db.password").get

// initialize JDBC driver & connection pool
Class.forName("org.postgresql.Driver")
ConnectionPool.singleton(s"jdbc:postgresql://$host:$port/$dbName", user, password)
ConnectionPool.dataSource().asInstanceOf[PoolingDataSource].setAccessToUnderlyingConnectionAllowed(true)

val connection = ConnectionPool.borrow()
val db: DB = DB(connection)

private val notificationsObservable: Observable[Array[PGNotification]] =
  Observable.interval(1 second, 500 millis)
    .map{ _ =>
      db.readOnly { implicit session =>
        val pgConnection = connection.asInstanceOf[DelegatingConnection].getInnermostDelegate.asInstanceOf[PGConnection]
        Option(pgConnection.getNotifications).getOrElse(Array[PGNotification]())
      }
    }
    .filter( _.nonEmpty )

notificationsObservable.subscribe {
  _.foreach { notification =>
    log.info(s"Notification received: Pid(${notification.getPID}) Channel(${notification.getName}) Payload(${notification.getParameter})")
  }
}