Закрытие потока Akka для очистки ресурсов

Question

При использовании потоков Akka, есть ли способ закрыть/закрыть поток, который больше не требуется для очистки ресурсов?

EDIT: Когда источник состоит из бесконечного потока, он никогда не может быть завершен, и я хотел бы остановить его до завершенного источника.

Пример использования:

Source.from(publisher) 
     .map((p) -> p) 
     .to(Sink.ignore()) 
     .run(materializer) 

Есть ли способ остановки потока?

Answer 1

Вы можете запустить Stream на независимой ActorMaterializer и вызвать остановку на ActorMaterializer по истечению определенного периода времени:

val actorSystem = ActorSystem() 

val temporaryStream = { 

    val localMat = ActorMaterializer()(actorSystem) 

    import actorSystem.dispatcher 
    actorSystem.scheduler.scheduleOnce(10 minutes) { localMat.shutdown() } 

    Source.from(publisher) 
     .map((p) -> p) 
     .to(Sink.ignore()) 
     .run()(localMat) 
} 

Точно так же вы можете вернуть ActorMaterializer вместо материализованного потока и выключений ActorMaterializer на основе некоторые внешние условия, кроме времени.