def withBroker()

in mqttv5/src/main/scala/org/apache/pekko/stream/connectors/mqttv5/settings.scala [280:473]
142 lines of code
1 McCabe index (conditional complexity)

  def withBroker(value: String): MqttConnectionSettings =
    copy(broker = value)

  def withClientId(value: String): MqttConnectionSettings =
    copy(clientId = value)

  def withPersistence(value: MqttClientPersistence): MqttConnectionSettings =
    copy(persistence = value)

  def withAuth(
      username: String,
      password: String
  ): MqttConnectionSettings =
    copy(auth = AuthSettings.Simple(username = username, password = password))

  /** Scala API */
  def withAuth(
      method: String,
      initialData: Array[Byte],
      authPacketHandler: (Int, MqttProperties) => (Int, MqttProperties)
  ): MqttConnectionSettings =
    copy(
      auth = AuthSettings.Enhanced(
        method = method,
        initialData = initialData,
        authPacketHandler = authPacketHandler
      )
    )

  /** Java API */
  def withAuth(
      method: String,
      initialData: Array[Byte],
      authPacketHandler: java.util.function.BiFunction[Int, MqttProperties, Pair[Int, MqttProperties]]
  ): MqttConnectionSettings =
    withAuth(
      method = method,
      initialData = initialData,
      authPacketHandler = (reasonCode: Int, properties: MqttProperties) =>
        authPacketHandler.apply(reasonCode, properties).toScala
    )

  def withSocketFactory(value: javax.net.ssl.SSLSocketFactory): MqttConnectionSettings =
    copy(socketFactory = Option(value))

  def withCleanStart(enabled: Boolean): MqttConnectionSettings =
    copy(cleanStart = CleanStartSettings(enabled = enabled, sessionExpiration = None))

  /** Scala API */
  def withCleanStart(enabled: Boolean, sessionExpiration: Option[FiniteDuration]): MqttConnectionSettings =
    copy(cleanStart = CleanStartSettings(enabled, sessionExpiration))

  /** Java API */
  def withCleanStart(enabled: Boolean, sessionExpiration: java.time.Duration): MqttConnectionSettings =
    copy(cleanStart = CleanStartSettings(enabled, Option(sessionExpiration).map(_.asScala)))

  def withWill(value: MqttMessage): MqttConnectionSettings =
    copy(will = Option(value))

  def withAutomaticReconnect(value: Boolean): MqttConnectionSettings =
    copy(automaticReconnect = value)

  /** Scala API */
  def withKeepAliveInterval(value: FiniteDuration): MqttConnectionSettings =
    copy(keepAliveInterval = value)

  /** Java API */
  def withKeepAliveInterval(value: java.time.Duration): MqttConnectionSettings =
    withKeepAliveInterval(value.asScala)

  /** Scala API */
  def withConnectionTimeout(value: FiniteDuration): MqttConnectionSettings =
    copy(connectionTimeout = value)

  /** Java API */
  def withConnectionTimeout(value: java.time.Duration): MqttConnectionSettings =
    withConnectionTimeout(value.asScala)

  /** Scala API */
  def withDisconnectQuiesceTimeout(value: FiniteDuration): MqttConnectionSettings =
    copy(disconnect = disconnect.copy(quiesceTimeout = value))

  /** Java API */
  def withDisconnectQuiesceTimeout(value: java.time.Duration): MqttConnectionSettings =
    withDisconnectQuiesceTimeout(value.asScala)

  /** Scala API */
  def withDisconnectTimeout(value: FiniteDuration): MqttConnectionSettings =
    copy(disconnect = disconnect.copy(timeout = value))

  /** Java API */
  def withDisconnectTimeout(value: java.time.Duration): MqttConnectionSettings =
    withDisconnectTimeout(value.asScala)

  def withServerUri(value: String): MqttConnectionSettings =
    copy(serverUris = Array(value))

  /** Scala API */
  def withServerUris(value: immutable.Seq[String]): MqttConnectionSettings =
    copy(serverUris = value.toArray)

  /** Java API */
  def withServerUris(value: java.util.List[String]): MqttConnectionSettings =
    copy(serverUris = value.asScala.toArray)

  def withSslHostnameVerifier(value: javax.net.ssl.HostnameVerifier): MqttConnectionSettings =
    copy(sslHostnameVerifier = Option(value))

  /** Scala API */
  def withSslProperties(value: Map[String, String]): MqttConnectionSettings =
    copy(sslProperties = value)

  /** Java API */
  def withSslProperties(value: java.util.Map[String, String]): MqttConnectionSettings =
    withSslProperties(value = value.asScala.toMap)

  def withOfflinePersistenceSettings(
      bufferSize: Int = 5000,
      deleteOldestMessage: Boolean = false,
      persistBuffer: Boolean = true
  ): MqttConnectionSettings =
    copy(
      offlinePersistence = Option(
        MqttOfflinePersistenceSettings(
          bufferSize = bufferSize,
          deleteOldestMessage = deleteOldestMessage,
          persistBuffer = persistBuffer
        )
      )
    )

  private def copy(
      broker: String = broker,
      clientId: String = clientId,
      persistence: MqttClientPersistence = persistence,
      cleanStart: CleanStartSettings = cleanStart,
      disconnect: DisconnectSettings = disconnect,
      offlinePersistence: Option[MqttOfflinePersistenceSettings] = offlinePersistence,
      auth: AuthSettings = auth,
      automaticReconnect: Boolean = automaticReconnect,
      keepAliveInterval: FiniteDuration = keepAliveInterval,
      connectionTimeout: FiniteDuration = connectionTimeout,
      serverUris: Array[String] = serverUris,
      will: Option[MqttMessage] = will,
      sslProperties: Map[String, String] = sslProperties,
      socketFactory: Option[javax.net.ssl.SSLSocketFactory] = socketFactory,
      sslHostnameVerifier: Option[javax.net.ssl.HostnameVerifier] = sslHostnameVerifier
  ): MqttConnectionSettings =
    new MqttConnectionSettings(
      broker = broker,
      clientId = clientId,
      persistence = persistence,
      disconnect = disconnect,
      offlinePersistence = offlinePersistence,
      auth = auth,
      cleanStart = cleanStart,
      automaticReconnect = automaticReconnect,
      keepAliveInterval = keepAliveInterval,
      connectionTimeout = connectionTimeout,
      serverUris = serverUris,
      will = will,
      sslProperties = sslProperties,
      socketFactory = socketFactory,
      sslHostnameVerifier = sslHostnameVerifier
    )

  override def toString: String =
    "MqttConnectionSettings(" +
    s"broker=$broker," +
    s"clientId=$clientId," +
    s"persistence=$persistence," +
    s"disconnect=$disconnect," +
    s"offlinePersistence=$offlinePersistence," +
    s"auth=${auth.asString}," +
    s"cleanStart=$cleanStart," +
    s"automaticReconnect=$automaticReconnect," +
    s"keepAliveInterval=$keepAliveInterval," +
    s"connectionTimeout=$connectionTimeout," +
    s"serverUris=$serverUris," +
    s"will=$will," +
    s"sslProperties=$sslProperties," +
    s"socketFactory=$socketFactory," +
    s"sslHostnameVerifier=$sslHostnameVerifier" +
    ")"
}

/**
 * Factory for connection settings passed to the underlying Paho client.
 *
 * Java docs for `MqttConnectionOptions` are not available;
 * see [[https://github.com/eclipse-paho/paho.mqtt.java/issues/1012]] or
 * [[org.eclipse.paho.mqttv5.client.MqttConnectionOptions]] for more info
 */
object MqttConnectionSettings {