Cleanup code by applying inspections

amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectionProvider.scala

@@ -143,11 +143,10 @@ final class AmqpDetailsConnectionProvider private (
  if (sslConfiguration.trustManager.isDefined)
  factory.useSslProtocol(sslConfiguration.protocol.get, sslConfiguration.trustManager.get)
  else factory.useSslProtocol(sslConfiguration.protocol.get)
- } else if (sslConfiguration.context.isDefined) {
+ } else if (sslConfiguration.context.isDefined)
  factory.useSslProtocol(sslConfiguration.context.get)
- } else {
+ else
  factory.useSslProtocol()
- }
  })
  requestedHeartbeat.foreach(factory.setRequestedHeartbeat)
  connectionTimeout.foreach(factory.setConnectionTimeout)
@@ -244,9 +243,8 @@ object AmqpCredentials {
 final class AmqpSSLConfiguration private (val protocol: Option[String] = None,
  val trustManager: Option[TrustManager] = None,
  val context: Option[SSLContext] = None) {
- if (protocol.isDefined && context.isDefined) {
+ if (protocol.isDefined && context.isDefined)
  throw new IllegalArgumentException("Protocol and context can't be defined in the same AmqpSSLConfiguration.")
- }
 
  def withProtocol(protocol: String): AmqpSSLConfiguration =
  copy(protocol = Some(protocol))
@@ -419,10 +417,8 @@ final class AmqpCachedConnectionProvider private (val provider: AmqpConnectionPr
  throw new ConcurrentModificationException(
  "Unexpected concurrent modification while closing the connection.")
  }
- } else {
- if (!state.compareAndSet(c, Connected(cachedConnection, clients - 1)))
- releaseRecursive(amqpConnectionProvider, connection)
- }
+ } else if (!state.compareAndSet(c, Connected(cachedConnection, clients - 1)))
+ releaseRecursive(amqpConnectionProvider, connection)
  case Closing => releaseRecursive(amqpConnectionProvider, connection)
  }
 

amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectorSettings.scala

@@ -152,7 +152,7 @@ final class TemporaryQueueSourceSettings private (
 }
 
 object TemporaryQueueSourceSettings {
- def apply(connectionProvider: AmqpConnectionProvider, exchange: String) =
+ def apply(connectionProvider: AmqpConnectionProvider, exchange: String): TemporaryQueueSourceSettings =
  new TemporaryQueueSourceSettings(connectionProvider, exchange)
 
  /**

amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AbstractAmqpAsyncFlowStageLogic.scala

@@ -72,9 +72,7 @@ import scala.concurrent.Promise
  val callback = getAsyncCallback[(DeliveryTag, Boolean)] {
  case (tag: DeliveryTag, multiple: Boolean) => confirmCallback(tag, multiple)
  }
- new ConfirmCallback { // cant use function literal because it doesn't work with 2.11
- override def handle(tag: DeliveryTag, multiple: Boolean): Unit = callback.invoke((tag, multiple))
- }
+ (tag: DeliveryTag, multiple: Boolean) => callback.invoke((tag, multiple))
  }
 
  private def onConfirmation(tag: DeliveryTag, multiple: Boolean): Unit = {
@@ -155,9 +153,8 @@ import scala.concurrent.Promise
  if (noAwaitingMessages && exitQueue.isEmpty) {
  streamCompletion.success(Done)
  super.onUpstreamFinish()
- } else {
+ } else
  log.debug("Received upstream finish signal - stage will be closed when all buffered messages are processed")
- }
 
  private def publish(message: WriteMessage): DeliveryTag = {
  val tag: DeliveryTag = channel.getNextPublishSeqNo
@@ -193,10 +190,9 @@ import scala.concurrent.Promise
 
  override protected def onTimer(timerKey: Any): Unit =
  timerKey match {
- case tag: DeliveryTag => {
+ case tag: DeliveryTag =>
  log.debug("Received timeout for deliveryTag {}.", tag)
  onRejection(tag, multiple = false)
- }
  case _ => ()
  }
 

amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpAsyncFlowStage.scala

@@ -58,15 +58,14 @@ import scala.concurrent.{ Future, Promise }
  buffer += (tag -> AwaitingMessage(tag, passThrough))
 
  override def dequeueAwaitingMessages(tag: DeliveryTag, multiple: Boolean): Iterable[AwaitingMessage[T]] =
- if (multiple) {
+ if (multiple)
  dequeueWhile((t, _) => t <= tag)
- } else {
+ else {
  setReady(tag)
- if (isAtHead(tag)) {
+ if (isAtHead(tag))
  dequeueWhile((_, message) => message.ready)
- } else {
+ else
  Seq.empty
- }
  }
 
  private def dequeueWhile(

amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpReplyToSinkStage.scala

@@ -31,7 +31,7 @@ import scala.concurrent.{ Future, Promise }
 private[amqp] final class AmqpReplyToSinkStage(replyToSinkSettings: AmqpReplyToSinkSettings)
  extends GraphStageWithMaterializedValue[SinkShape[WriteMessage], Future[Done]] { stage =>
 
- val in = Inlet[WriteMessage]("AmqpReplyToSink.in")
+ val in: Inlet[WriteMessage] = Inlet[WriteMessage]("AmqpReplyToSink.in")
 
  override def shape: SinkShape[WriteMessage] = SinkShape.of(in)
 
@@ -82,9 +82,8 @@ private[amqp] final class AmqpReplyToSinkStage(replyToSinkSettings: AmqpReplyToS
  elem.immediate,
  elem.properties.orNull,
  elem.bytes.toArray)
- } else if (settings.failIfReplyToMissing) {
+ } else if (settings.failIfReplyToMissing)
  onFailure(new RuntimeException("Reply-to header was not set"))
- }
 
  tryPull(in)
  }

amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpRpcFlowStage.scala

@@ -42,8 +42,8 @@ private[amqp] final class AmqpRpcFlowStage(writeSettings: AmqpWriteSettings, buf
  extends GraphStageWithMaterializedValue[FlowShape[WriteMessage, CommittableReadResult], Future[String]] {
  stage =>
 
- val in = Inlet[WriteMessage]("AmqpRpcFlow.in")
- val out = Outlet[CommittableReadResult]("AmqpRpcFlow.out")
+ val in: Inlet[WriteMessage] = Inlet[WriteMessage]("AmqpRpcFlow.in")
+ val out: Outlet[CommittableReadResult] = Outlet[CommittableReadResult]("AmqpRpcFlow.out")
 
  override def shape: FlowShape[WriteMessage, CommittableReadResult] = FlowShape.of(in, out)
 
@@ -70,7 +70,7 @@ private[amqp] final class AmqpRpcFlowStage(writeSettings: AmqpWriteSettings, buf
  val consumerCallback = getAsyncCallback(handleDelivery)
 
  val commitCallback = getAsyncCallback[AckArguments] {
- case AckArguments(deliveryTag, multiple, promise) => {
+ case AckArguments(deliveryTag, multiple, promise) =>
  try {
  channel.basicAck(deliveryTag, multiple)
  unackedMessages -= 1
@@ -81,10 +81,9 @@ private[amqp] final class AmqpRpcFlowStage(writeSettings: AmqpWriteSettings, buf
  } catch {
  case e: Throwable => promise.failure(e)
  }
- }
  }
  val nackCallback = getAsyncCallback[NackArguments] {
- case NackArguments(deliveryTag, multiple, requeue, promise) => {
+ case NackArguments(deliveryTag, multiple, requeue, promise) =>
  try {
  channel.basicNack(deliveryTag, multiple, requeue)
  unackedMessages -= 1
@@ -95,7 +94,6 @@ private[amqp] final class AmqpRpcFlowStage(writeSettings: AmqpWriteSettings, buf
  } catch {
  case e: Throwable => promise.failure(e)
  }
- }
  }
 
  val amqpSourceConsumer = new DefaultConsumer(channel) {
@@ -105,7 +103,7 @@ private[amqp] final class AmqpRpcFlowStage(writeSettings: AmqpWriteSettings, buf
  body: Array[Byte]): Unit =
  consumerCallback.invoke(
  new CommittableReadResult {
- override val message = ReadResult(ByteString(body), envelope, properties)
+ override val message: ReadResult = ReadResult(ByteString(body), envelope, properties)
 
  override def ack(multiple: Boolean): Future[Done] = {
  val promise = Promise[Done]()
@@ -148,21 +146,19 @@ private[amqp] final class AmqpRpcFlowStage(writeSettings: AmqpWriteSettings, buf
  }
 
  def handleDelivery(message: CommittableReadResult): Unit =
- if (isAvailable(out)) {
+ if (isAvailable(out))
  pushMessage(message)
- } else if (queue.size + 1 > bufferSize) {
+ else if (queue.size + 1 > bufferSize)
  onFailure(new RuntimeException(s"Reached maximum buffer size $bufferSize"))
- } else {
+ else
  queue.enqueue(message)
- }
 
  setHandler(
  out,
  new OutHandler {
  override def onPull(): Unit =
- if (queue.nonEmpty) {
+ if (queue.nonEmpty)
  pushMessage(queue.dequeue())
- }
 
  override def onDownstreamFinish(cause: Throwable): Unit = {
  setKeepGoing(true)
@@ -207,15 +203,14 @@ private[amqp] final class AmqpRpcFlowStage(writeSettings: AmqpWriteSettings, buf
 
  val expectedResponses: Int = {
  val headers = props.getHeaders
- if (headers == null) {
+ if (headers == null)
  responsesPerMessage
- } else {
+ else {
  val r = headers.get("expectedReplies")
- if (r != null) {
+ if (r != null)
  r.asInstanceOf[Int]
- } else {
+ else
  responsesPerMessage
- }
  }
  }
 

amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpSourceStage.scala

@@ -94,9 +94,8 @@ private[amqp] final class AmqpSourceStage(settings: AmqpSourceSettings, bufferSi
  properties: BasicProperties,
  body: Array[Byte]): Unit = {
  val message = if (ackRequired) {
-
  new CommittableReadResult {
- override val message = ReadResult(ByteString(body), envelope, properties)
+ override val message: ReadResult = ReadResult(ByteString(body), envelope, properties)
 
  override def ack(multiple: Boolean): Future[Done] = {
  val promise = Promise[Done]()
@@ -155,21 +154,19 @@ private[amqp] final class AmqpSourceStage(settings: AmqpSourceSettings, bufferSi
  }
 
  def handleDelivery(message: CommittableReadResult): Unit =
- if (isAvailable(out)) {
+ if (isAvailable(out))
  pushMessage(message)
- } else if (queue.size + 1 > bufferSize) {
+ else if (queue.size + 1 > bufferSize)
  onFailure(new RuntimeException(s"Reached maximum buffer size $bufferSize"))
- } else {
+ else
  queue.enqueue(message)
- }
 
  setHandler(
  out,
  new OutHandler {
  override def onPull(): Unit =
- if (queue.nonEmpty) {
+ if (queue.nonEmpty)
  pushMessage(queue.dequeue())
- }
 
  override def onDownstreamFinish(cause: Throwable): Unit =
  if (unackedMessages == 0) super.onDownstreamFinish(cause)

amqp/src/test/scala/docs/scaladsl/AmqpDocsSpec.scala

@@ -35,7 +35,7 @@ class AmqpDocsSpec extends AmqpSpec {
 
  override implicit val patienceConfig: PatienceConfig = PatienceConfig(10.seconds)
 
- val businessLogic: CommittableReadResult => Future[CommittableReadResult] = Future.successful(_)
+ val businessLogic: CommittableReadResult => Future[CommittableReadResult] = Future.successful
 
  "The AMQP Connectors" should {
 
@@ -158,7 +158,7 @@ class AmqpDocsSpec extends AmqpSpec {
  val mergingFlow = mergedSources
  .viaMat(KillSwitches.single)(Keep.right)
  .to(Sink.fold(Set.empty[Int]) {
- case (seen, (branch, element)) =>
+ case (seen, (branch, _)) =>
  if (seen.size == fanoutSize) completion.trySuccess(Done)
  seen + branch
  })

amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/AmqpProxyConnection.scala

@@ -29,21 +29,21 @@ import com.rabbitmq.client._
  * otherwise undefined
  */
 class AmqpProxyConnection(protected val delegate: Connection) extends Connection {
- override def getAddress: InetAddress = delegate.getAddress()
+ override def getAddress: InetAddress = delegate.getAddress
 
- override def getPort: Int = delegate.getPort()
+ override def getPort: Int = delegate.getPort
 
- override def getChannelMax: Int = delegate.getChannelMax()
+ override def getChannelMax: Int = delegate.getChannelMax
 
- override def getFrameMax: Int = delegate.getFrameMax()
+ override def getFrameMax: Int = delegate.getFrameMax
 
- override def getHeartbeat: Int = delegate.getHeartbeat()
+ override def getHeartbeat: Int = delegate.getHeartbeat
 
- override def getClientProperties: util.Map[String, AnyRef] = delegate.getClientProperties()
+ override def getClientProperties: util.Map[String, AnyRef] = delegate.getClientProperties
 
- override def getClientProvidedName: String = delegate.getClientProvidedName()
+ override def getClientProvidedName: String = delegate.getClientProvidedName
 
- override def getServerProperties: util.Map[String, AnyRef] = delegate.getServerProperties()
+ override def getServerProperties: util.Map[String, AnyRef] = delegate.getServerProperties
 
  override def createChannel(): Channel = delegate.createChannel()
 
@@ -76,9 +76,9 @@ class AmqpProxyConnection(protected val delegate: Connection) extends Connection
 
  override def clearBlockedListeners(): Unit = delegate.clearBlockedListeners()
 
- override def getExceptionHandler: ExceptionHandler = delegate.getExceptionHandler()
+ override def getExceptionHandler: ExceptionHandler = delegate.getExceptionHandler
 
- override def getId: String = delegate.getId()
+ override def getId: String = delegate.getId
 
  override def setId(s: String): Unit = delegate.setId(s)
 
@@ -88,9 +88,9 @@ class AmqpProxyConnection(protected val delegate: Connection) extends Connection
  override def removeShutdownListener(shutdownListener: ShutdownListener): Unit =
  delegate.removeShutdownListener(shutdownListener)
 
- override def getCloseReason: ShutdownSignalException = delegate.getCloseReason()
+ override def getCloseReason: ShutdownSignalException = delegate.getCloseReason
 
  override def notifyListeners(): Unit = delegate.notifyListeners()
 
- override def isOpen: Boolean = delegate.isOpen()
+ override def isOpen: Boolean = delegate.isOpen
 }

amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpGraphStageLogicConnectionShutdownSpec.scala

@@ -55,7 +55,7 @@ class AmqpGraphStageLogicConnectionShutdownSpec
  val shutdownsAdded = new AtomicInteger()
  val shutdownsRemoved = new AtomicInteger()
 
- override def beforeEach() = {
+ override def beforeEach(): Unit = {
  shutdownsAdded.set(0)
  shutdownsRemoved.set(0)
  }

avroparquet/src/main/scala/org/apache/pekko/stream/connectors/avroparquet/impl/AvroParquetFlow.scala

@@ -41,7 +41,6 @@ private[avroparquet] class AvroParquetFlow[T <: GenericRecord](writer: ParquetWr
  new InHandler {
 
  override def onUpstreamFinish(): Unit =
- // super.onUpstreamFinish()
  completeStage()
 
  override def onUpstreamFailure(ex: Throwable): Unit = {

avroparquet/src/test/scala/docs/scaladsl/AvroParquetFlowSpec.scala

@@ -71,7 +71,7 @@ class AvroParquetFlowSpec
  val n: Int = 2
  val file: String = genFinalFile.sample.get
  val documents: List[Document] = genDocuments(n).sample.get
- val avroDocuments: List[Record] = documents.map(format.to(_))
+ val avroDocuments: List[Record] = documents.map(format.to)
  val writer: ParquetWriter[Record] = parquetWriter[Record](file, conf, schema)
 
  // when

avroparquet/src/test/scala/docs/scaladsl/AvroParquetSinkSpec.scala

@@ -63,7 +63,7 @@ class AvroParquetSinkSpec
  val documents: List[Document] = genDocuments(n).sample.get
  val writer: ParquetWriter[Record] = parquetWriter[Record](file, conf, schema)
  // #init-sink
- val records: List[Record] = documents.map(format.to(_))
+ val records: List[Record] = documents.map(format.to)
  val source: Source[Record, NotUsed] = Source(records)
  val result: Future[Done] = source
  .runWith(AvroParquetSink(writer))

avroparquet/src/test/scala/docs/scaladsl/AvroParquetSourceSpec.scala

@@ -69,7 +69,7 @@ class AvroParquetSourceSpec
  val n: Int = 4
  val file: String = genFinalFile.sample.get
  val documents: List[Document] = genDocuments(n).sample.get
- val avroDocuments: List[Record] = documents.map(format.to(_))
+ val avroDocuments: List[Record] = documents.map(format.to)
  Source(avroDocuments)
  .toMat(AvroParquetSink(parquetWriter(file, conf, schema)))(Keep.right)
  .run()

aws-event-bridge/src/main/scala/org/apache/pekko/stream/connectors/aws/eventbridge/EventBridgePublishSettings.scala

@@ -27,7 +27,7 @@ final class EventBridgePublishSettings private (val concurrency: Int) {
 
  def withConcurrency(concurrency: Int): EventBridgePublishSettings = copy(concurrency = concurrency)
 
- def copy(concurrency: Int) = new EventBridgePublishSettings(concurrency)
+ def copy(concurrency: Int): EventBridgePublishSettings = new EventBridgePublishSettings(concurrency)
 
  override def toString: String =
  "EventBridgePublishSettings(" +

aws-event-bridge/src/test/scala/org/apache/pekko/stream/connectors/aws/eventbridge/EventBridgePublishMockSpec.scala

@@ -32,7 +32,7 @@ class EventBridgePublishMockSpec extends AnyFlatSpec with DefaultTestContext wit
 
  private def entryDetail(detail: String, eventBusName: Option[String] = None): PutEventsRequestEntry = {
  val entry = PutEventsRequestEntry.builder().detail(detail)
- eventBusName.map(entry.eventBusName(_))
+ eventBusName.map(entry.eventBusName)
  entry.build()
  }
 
@@ -129,7 +129,7 @@ class EventBridgePublishMockSpec extends AnyFlatSpec with DefaultTestContext wit
  }
 
  it should "fail stage if upstream failure occurs" in {
- case class MyCustomException(message: String) extends Exception(message)
+ final case class MyCustomException(message: String) extends Exception(message)
 
  val (probe, future) =
  TestSource.probe[Seq[PutEventsRequestEntry]].via(EventBridgePublisher.flowSeq()).toMat(Sink.seq)(Keep.both).run()