A Pub/Sub subscriber wrapper that supports local emulator

SubscriberWrapper.scala

class SubscriberWrapper(subscriber: Subscriber) extends AutoCloseable {
  override def close(): Unit = subscriber.stopAsync()
  def startAsync: ApiService = subscriber.startAsync()
  def awaitTerminated(): Unit = subscriber.awaitTerminated()
}

object SubscriberWrapper {
  def apply(projectId: String, subscription: String, localPubSubEndpoint: Option[String], receiver: MessageReceiver): SubscriberWrapper = {
    val baseBuilder = Subscriber.newBuilder(ProjectSubscriptionName.format(projectId, subscription), receiver)
    val subscriber =
      localPubSubEndpoint
        .map(h => handleLocalExecution(baseBuilder, h))
        .map(_.build())
        .getOrElse(baseBuilder.build())
    new SubscriberWrapper(subscriber)
  }

  private def handleLocalExecution(baseBuilder: Subscriber.Builder, localPubSubEndpoint: String): Subscriber.Builder = {
    val channel = ManagedChannelBuilder
      .forTarget(localPubSubEndpoint)
      .usePlaintext()
      .asInstanceOf[ManagedChannelBuilder[NettyChannelBuilder]]
      .build()
    baseBuilder
      .setChannelProvider(FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel)))
      .setCredentialsProvider(NoCredentialsProvider.create)
  }

}