如果下行仍在工作，如何在Akka流中丢弃下行消息

import akka.stream.scaladsl._

val source: Source[Message, NotUsed] = ???
val sink: Sink[Message, Future[Done]] = ???

val filteredSource: Source[Message, NotUsed] = source.filter { message =>
  // 根据特定的条件判断是否需要丢弃下行消息
  // 返回true表示保留消息，返回false表示丢弃消息
  // 例如，可以根据消息的类型或内容进行判断
  // 如果需要丢弃消息，可以记录日志或进行其他处理
  false
}

val stream: RunnableGraph[NotUsed] = filteredSource.to(sink)

import akka.stream.scaladsl._

val source: Source[Message, NotUsed] = ???
val sink: Sink[Message, Future[Done]] = ???

val processedSource: Source[Message, NotUsed] = source.collect {
  case message if shouldProcess(message) =>
    // 对需要处理的消息进行处理
    // 例如，可以对消息进行转换、过滤或其他操作
    // 如果不需要处理消息，可以返回原始消息或空值
    message
}

val stream: RunnableGraph[NotUsed] = processedSource.to(sink)