Fix NPE

pbernet · May 22, 2024 · 4be426b · 4be426b
1 parent c5fcb6e
commit 4be426b
Showing 1 changed file with 4 additions and 5 deletions.
diff --git a/src/main/scala/alpakka/tcp_to_websockets/websockets/Kafka2SSE.scala b/src/main/scala/alpakka/tcp_to_websockets/websockets/Kafka2SSE.scala
@@ -17,7 +17,6 @@ import org.apache.pekko.stream.scaladsl.{Keep, RestartSource, Sink, Source}
 import org.slf4j.{Logger, LoggerFactory}
 
 import java.util.Locale
-import scala.concurrent.Future
 import scala.concurrent.duration.*
 import scala.language.postfixOps
 import scala.util.{Failure, Success}
@@ -47,10 +46,10 @@ class Kafka2SSE(mappedPortKafka: Int = 9092) {
     clientKillSwitch = backoffClient(address, port)
   }
 
-  def stop(): Future[Http.HttpTerminated] = {
+  def stop(): Unit = {
     logger.info("Stopping...")
     clientKillSwitch.shutdown()
-    serverBinding.terminate(10.seconds)
+    if (serverBinding != null) serverBinding.terminate(10.seconds) else {}
   }
 
   private def createConsumerSettings(group: String): ConsumerSettings[String, String] = {
@@ -91,10 +90,10 @@ class Kafka2SSE(mappedPortKafka: Int = 9092) {
     val bindingFuture = Http().newServerAt(address, port).bindFlow(route)
     bindingFuture.onComplete {
       case Success(binding) =>
-        logger.info("Server started, listening on: " + binding.localAddress)
+        logger.info(s"Server started, listening on: {}", binding.localAddress)
         serverBinding = binding
       case Failure(e) =>
-        logger.info(s"Server could not bind to $address:$port. Exception message: ${e.getMessage}")
+        logger.info(s"Server could not bind to: $address:$port. Exception message: ${e.getMessage}")
         system.terminate()
     }
   }