How to detect that a consumer has disconnected from a stream

Akka Akka Streams & Alpakka
, ,
1

Given the following

    val ((queue, ks), source) =
      Source
        .queue[MyThing](queueSize, OverflowStrategy.dropHead)
        .viaMat(KillSwitches.single)(Keep.both)
        .preMaterialize()

How can I detect when the consumer of the source has disconnected so that I can clean up any resources that were created to serve this stream?

I tried adding

stream.watchTermination() { (_, done) =>
      done.onComplete { _ =>
        // cleanup
      }
      NotUsed
    }

but it is never invoked

2

Actually - cancel that. When I use watchTermination correctly, it does work. :doh:

1 Like