mqtt-streaming: Fix a memory leak if subscribe/unsubscribe fails

akka · Apr 4, 2019 · 1994542 · 1994542
1 parent ede7e69
commit 1994542
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ServerState.scala b/mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ServerState.scala
@@ -880,6 +880,7 @@ import scala.util.{Failure, Success}
  )
  case UnobtainablePacketId =>
  data.local.failure(SubscribeFailed)
+ data.subscribed.failure(SubscribeFailed)
  throw SubscribeFailed
  }
 
@@ -895,6 +896,7 @@ import scala.util.{Failure, Success}
  data.subscribed.success(remote)
  Behaviors.stopped
  case ReceiveSubAckTimeout =>
+ data.subscribed.failure(SubscribeFailed)
  throw SubscribeFailed
  }
  .receiveSignal {
@@ -977,6 +979,7 @@ import scala.util.{Failure, Success}
  )
  case UnobtainablePacketId =>
  data.local.failure(UnsubscribeFailed)
+ data.unsubscribed.failure(UnsubscribeFailed)
  throw UnsubscribeFailed
  }
  }
@@ -992,6 +995,7 @@ import scala.util.{Failure, Success}
  data.unsubscribed.success(Done)
  Behaviors.stopped
  case ReceiveUnsubAckTimeout =>
+ data.unsubscribed.failure(UnsubscribeFailed)
  throw UnsubscribeFailed
  }
  .receiveSignal {