/*
 * Copyright (C) 2014-2020 Lightbend Inc. <https://www.lightbend.com>
 */

package akka.stream.scaladsl

import akka.stream._
import akka.stream.stage._

trait ZipWithApply {

  [2..22#/**
   * Create a new `ZipWith` specialized for 1 inputs.
   *
   * @param zipper zipping-function from the input values to the output value
   */
  def apply[[#A1#], O](zipper: ([#A1#]) => O): ZipWith1[[#A1#], O] =
    new ZipWith1(zipper)
  #

  ]

}

[2..22#/** `ZipWith` specialized for 1 inputs */
class ZipWith1[[#A1#], O] (val zipper: ([#A1#]) => O) extends GraphStage[FanInShape1[[#A1#], O]] {
  override def initialAttributes = Attributes.name("ZipWith1")
  override val shape: FanInShape1[[#A1#], O] = new FanInShape1[[#A1#], O]("ZipWith1")
  def out: Outlet[O] = shape.out
  [#val in0: Inlet[A1] = shape.in0#
  ]

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) {
    var pending = ##0
    // Without this field the completion signalling would take one extra pull
    var willShutDown = false

    private def pushAll(): Unit = {
      push(out, zipper([#grab(in0)#]))
      if (willShutDown) completeStage()
      else {
        [#pull(in0)#
        ]
      }
    }

    override def preStart(): Unit = {
      [#pull(in0)#
      ]
    }

    [#setHandler(in0, new InHandler {
      override def onPush(): Unit = {
        pending -= ##1
        if (pending == ##0) pushAll()
      }

      override def onUpstreamFinish(): Unit = {
        if (!isAvailable(in0)) completeStage()
        willShutDown = true
      }

    })#
    ]

    setHandler(out, new OutHandler {
      override def onPull(): Unit = {
        pending += shape.inlets.size
        if (pending == ##0) pushAll()
      }
    })
  }

  override def toString = "ZipWith1"

}
#
]
