akka-streams - 从应用角度学习：basic stream parts

  def maybe[T]: Source[T, Promise[Option[T]]] =
    Source.fromGraph(MaybeSource.asInstanceOf[Graph[SourceShape[T], Promise[Option[T]]]])

Source(1 to 10).async.via(Flow[Int].map(i => i + 1)).async.runWith(sink)

// A source that can be signalled explicitly from the outside
val source: Source[Int, Promise[Option[Int]]] = Source.maybe[Int]

// A flow that internally throttles elements to 1/second, and returns a Cancellable
// which can be used to shut down the stream
val flow: Flow[Int, Int, Cancellable] = throttler

// A sink that returns the first element of a stream in the returned Future
val sink: Sink[Int, Future[Int]] = Sink.head[Int]

val stream: RunnableGraph[(Cancellable, Future[Int])] =
  source.viaMat(flow)(Keep.right).toMat(sink)(Keep.both)

val stream1: RunnableGraph[(Promise[Option[Int]], Cancellable, Future[Int])] =
  source.viaMat(flow)(Keep.both).toMat(sink)(Keep.both)

// Using runWith will always give the materialized values of the stages added
// by runWith() itself
val r4: Future[Int] = source.via(flow).runWith(sink)
val r5: Promise[Option[Int]] = flow.to(sink).runWith(source)
val r6: (Promise[Option[Int]], Future[Int]) = flow.runWith(source, sink)

def filter(p: Out => Boolean): Repr[Out] = via(Filter(p))

@InternalApi private[akka] final case class Filter[T](p: T => Boolean) extends SimpleLinearGraphStage[T] {
  override def initialAttributes: Attributes = DefaultAttributes.filter

  override def toString: String = "Filter"

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) with OutHandler with InHandler {
      def decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider

      private var buffer: OptionVal[T] = OptionVal.none

      override def preStart(): Unit = pull(in)
      override def onPush(): Unit =
        try {
          val elem = grab(in)
          if (p(elem))
            if (isAvailable(out)) {
              push(out, elem)
              pull(in)
            } else
              buffer = OptionVal.Some(elem)
          else pull(in)
        } catch {
          case NonFatal(ex) =>
            decider(ex) match {
              case Supervision.Stop => failStage(ex)
              case _                => pull(in)
            }
        }

      override def onPull(): Unit =
        buffer match {
          case OptionVal.Some(value) =>
            push(out, value)
            buffer = OptionVal.none
            if (!isClosed(in)) pull(in)
            else completeStage()
          case _ => // already pulled
        }

      override def onUpstreamFinish(): Unit =
        if (buffer.isEmpty) super.onUpstreamFinish()
      // else onPull will complete

      setHandlers(in, out, this)
    }
}

 def getVouchers(terminalid: String, susp: Boolean)(implicit classicSystem: ActorSystem) = {
    implicit val session = CassandraSessionRegistry(classicSystem).sessionFor("alpakka.cassandra")
    implicit val ec = classicSystem.dispatcher
    var stmt = "select * from pos_on_cloud.txn_log where terminal = ? and txndate = ?"
    if (susp) stmt = "select * from pos_on_cloud.txn_hold where terminal = ? and txndate = ?"
    val source  = session.select(stmt,terminalid,LocalDate.now.format(DateTimeFormatter.ofPattern("yyyyMMdd")))
    val sink = Sink.fold[List[TxnItem],TxnItem](List[TxnItem]()){(acc,txn) =>
      if (acc.isEmpty) txn.copy(price = 1) :: acc
      else
        if (acc.head.num == txn.num) {
          if (txn.salestype == SALESTYPE.itm &&
            txn.txntype == TXNTYPE.sales) {
            val nacc = acc.head.copy(
              price = acc.head.price + 1,
              qty = acc.head.qty + txn.qty,
              amount = acc.head.amount + txn.amount,
              dscamt = acc.head.dscamt + txn.dscamt
            )
            nacc :: acc.drop(1)
          } else acc
        }
        else txn :: acc
    }
    for {
      vchs <- source.map(TxnItem.fromCqlRow).toMat(sink)(Keep.right).run()
      _ <- session.close(ec)
    } yield vchs
  }

    def futTxns(items: Seq[TxnItem]): Future[Seq[TxnItem]] = Source(items.toSeq)
      .via(
        CassandraFlow.create(CassandraWriteSettings.defaults,
          CQLScripts.insertTxns,
          statementBinder)
      )
      .runWith(Sink.seq)