//设置package包名称以及导入依赖的类
package main.scala

import scala.collection.mutable.ArrayBuffer

import spray.json.DefaultJsonProtocol
import spray.json.DeserializationException
import spray.json.JsArray
import spray.json.JsNumber
import spray.json.JsObject
import spray.json.JsString
import spray.json.JsValue
import spray.json.JsonFormat
import spray.json.pimpAny

trait JsonFormats extends DefaultJsonProtocol {
  case class SendTweet(userId: Int, time: Long, msg: String)
  case class UserProfile(id: Int, name: String, statusCount: Int, favoritesCount: Int, followersCount: Int, followingCount: Int)
  case class SendMsg(senderId: Int, time: Long, msg: String, recepientId: Int)

  implicit val tweetFormat = jsonFormat3(SendTweet)
  implicit val userProfileFormat = jsonFormat6(UserProfile)
  implicit val reTweetFormat = jsonFormat4(SendMsg)

  implicit object TimelineJsonFormat extends JsonFormat[Project4Server.Tweets] {
    def write(c: Project4Server.Tweets) = JsObject(
      "authorId" -> JsNumber(c.authorId),
      "message" -> JsString(c.message),
      "timeStamp" -> JsString(c.timeStamp.toString),
      "tweetId" -> JsString(c.tweetId),
      "mentions" -> JsArray(c.mentions.map(_.toJson).toVector),
      "hashTags" -> JsArray(c.hashtags.map(_.toJson).toVector))
    def read(value: JsValue) = {
      value.asJsObject.getFields("tweetId", "authorId", "message", "timeStamp", "mentions", "hashTags") match {
        case Seq(JsString(tweetId), JsNumber(authorId), JsString(message), JsString(timeStamp), JsArray(mentions), JsArray(hashTags)) =>
          new Project4Server.Tweets(tweetId, authorId.toInt, message, timeStamp.toLong, mentions.map(_.convertTo[String]).to[ArrayBuffer], hashTags.map(_.convertTo[String]).to[ArrayBuffer])
        case _ => throw new DeserializationException("Tweets expected")
      }
    }
  }

  implicit object MessagesJsonFormat extends JsonFormat[Project4Server.Messages] {
    def write(c: Project4Server.Messages) = JsObject(
      "authorId" -> JsNumber(c.authorId),
      "message" -> JsString(c.message),
      "timeStamp" -> JsString(c.timeStamp.toString),
      "tweetId" -> JsString(c.tweetId),
      "mentions" -> JsArray(c.mentions.map(_.toJson).toVector),
      "hashTags" -> JsArray(c.hashtags.map(_.toJson).toVector),
      "recepientId" -> JsNumber(c.recepientId))
    def read(value: JsValue) = {
      value.asJsObject.getFields("tweetId", "authorId", "message", "timeStamp", "mentions", "hashTags", "recepientId") match {
        case Seq(JsString(tweetId), JsNumber(authorId), JsString(message), JsString(timeStamp), JsArray(mentions), JsArray(hashTags), JsNumber(recepientId)) =>
          new Project4Server.Messages(recepientId.toInt, tweetId, authorId.toInt, message, timeStamp.toLong, mentions.map(_.convertTo[String]).to[ArrayBuffer], hashTags.map(_.convertTo[String]).to[ArrayBuffer])
        case _ => throw new DeserializationException("Tweets expected")
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.durooma.api.route

import java.sql.Date

import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import com.durooma.api.model._
import com.durooma.db.Tables
import org.joda.time.DateTime
import org.joda.time.format.{DateTimeFormatter, ISODateTimeFormat}
import spray.json.{DefaultJsonProtocol, DeserializationException, JsString, JsValue, RootJsonFormat}


trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol {

  implicit object DateJsonFormat extends RootJsonFormat[DateTime] {

    private val parserISO : DateTimeFormatter = ISODateTimeFormat.dateTimeNoMillis()

    override def write(obj: DateTime) = JsString(parserISO.print(obj))

    override def read(json: JsValue) : DateTime = json match {
      case JsString(s) => parserISO.parseDateTime(s)
      case _ => throw DeserializationException("Invalid date format: " + json)
    }
  }

  implicit object SqlDateJsonFormat extends RootJsonFormat[Date] {

    override def write(obj: Date) = JsString(obj.toString)
    override def read(json: JsValue) = json match {
      case JsString(s) => Date.valueOf(s)
      case _ => throw DeserializationException("Invalid date format: " + json)
    }

  }

  implicit val userFormat = jsonFormat5(User.apply)
  implicit val userRegistrationFormat = jsonFormat5(UserRegistration.apply)
  implicit val accountFormat = jsonFormat4(Account.apply)
  implicit val accounBodyFormat = jsonFormat2(AccountBody.apply)
  implicit val labelFormat = jsonFormat3(Tables.LabelRow.apply)
  implicit val transactionFormat = jsonFormat8(Transaction.apply)
  implicit val transactionBodyFormat = jsonFormat7(TransactionBody.apply)
  implicit val sessionFormat = jsonFormat3(Session.apply)
  implicit val credentialsFormat = jsonFormat2(CustomCredentials.apply)

} 

//设置package包名称以及导入依赖的类
package com.github.chaabaj.openid

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpRequest
import akka.stream.ActorMaterializer
import com.github.chaabaj.openid.exceptions.{MalformedResponseException, WebServiceException}
import com.github.chaabaj.openid.utils.JsonResponseParser
import spray.json.JsValue

import scala.concurrent.duration.FiniteDuration
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}

private [openid] class HttpClient(implicit actorSystem: ActorSystem, timeout: FiniteDuration) {
  implicit val materializer = ActorMaterializer()
  val responseParser = new JsonResponseParser
  val http = Http()

  def request(httpRequest: HttpRequest)(implicit exc: ExecutionContext): Future[JsValue] =
    for {
      response <- http.singleRequest(httpRequest)
      body <- response.entity.toStrict(timeout).map(_.data.utf8String)
      data <- {
        responseParser.parse(body) match {
          case Success(data) =>
            if (response.status.isFailure()) {
              Future.failed(WebServiceException(response.status, data))
            } else {
              Future.successful(data)
            }
          case Failure(ex) => Future.failed(MalformedResponseException(response.status, ex.toString))
        }
      }
    } yield data
}

private[openid] object HttpClient {
  def apply()(implicit system: ActorSystem, _timeout: FiniteDuration): HttpClient =
    new HttpClient()
} 

//设置package包名称以及导入依赖的类
package com.jjabuk.bookstore.catalog.protocols

import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import com.jjabuk.bookstore.catalog.protocols.CatalogueProtocol.{Book, BookAdded, BooksFound}
import reactivemongo.bson.BSONObjectID
import spray.json.{DefaultJsonProtocol, JsArray, JsObject, JsString, JsValue, RootJsonFormat}

trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol {

  implicit val BookAddedFormat = jsonFormat1(BookAdded.apply)

  implicit object BooksFoundFormat extends RootJsonFormat[BooksFound] {
    override def read(json: JsValue): BooksFound = ???

    override def write(b: BooksFound): JsValue = JsObject(
      "books" -> JsArray(b.books.map(book => BookFormat.write(book)).toVector)
    )
  }

  implicit object BookFormat extends RootJsonFormat[Book] {
    override def read(value: JsValue) = {
      val uuid = fromField[Option[String]](value, "uuid")
      val isbn = fromField[String](value, "isbn")
      val title = fromField[String](value, "title")
      val review = fromField[Option[String]](value, "review")
      val publisher = fromField[Option[String]](value, "publisher")
      Book(uuid.getOrElse(BSONObjectID.generate().stringify), isbn, title, review, publisher)
    }

    override def write(obj: Book): JsValue = JsObject(
      "uuid" -> JsString(obj.uuid),
      "isbn" -> JsString(obj.isbn),
      "title" -> JsString(obj.title),
      "review" -> JsString(obj.review.getOrElse("")),
      "publisher" -> JsString(obj.publisher.getOrElse(""))
    )
  }
} 

//设置package包名称以及导入依赖的类
package proton.users

import java.io.{PrintWriter, StringWriter}
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util.UUID

import spray.json.{JsValue, JsonFormat, _}

trait UsersProtocol {
  private def getStackTrace(t: Throwable) = {
    val sw: StringWriter = new StringWriter()
    val pw: PrintWriter = new PrintWriter(sw)
    t.printStackTrace(pw)
    sw.toString
  }

  implicit object ThrowableWriter extends RootJsonWriter[Throwable] {
    def write(t: Throwable) = JsObject(
      "message" -> JsString(t.getMessage),
      "cause" -> t.getCause.toJson,
      "stackTrace" -> JsString(getStackTrace(t))
    )
  }

  implicit object MessageFormat extends RootJsonWriter[Message] {
    def write(m: Message) = JsObject(
      "summary" -> JsString(m.summary),
      "errorCode" -> JsNumber(m.errorCode)
    )
  }

  implicit object ValidationFormat extends RootJsonWriter[Validation] {
    def write(v: Validation) = {
      val fields = Seq[Option[JsField]](
        Some("message"   -> JsString(v.message)),
        Some("errorCode" -> JsNumber(v.errorCode)),
        v.exception.map(exception => "exception" -> exception.toJson)
      )
      JsObject(fields.flatten: _*)
    }
  }

  implicit object UUIDFormat extends JsonFormat[UUID] {
    def write(uuid: UUID) = JsString(uuid.toString)
    def read(value: JsValue) = value match {
      case JsString(uuid) => UUID.fromString(uuid)
      case _ => deserializationError("UUID expected.")
    }
  }

  implicit object LocalDateTimeFormat extends JsonFormat[LocalDateTime] {
    def write(dateTime: LocalDateTime) = JsString(dateTime.format(DateTimeFormatter.ISO_LOCAL_DATE_TIME))
    def read(value: JsValue) = value match {
      case JsString(dateTime) => LocalDateTime.parse(dateTime, DateTimeFormatter.ISO_LOCAL_DATE_TIME)
      case _ => deserializationError("LocalDateTime expected.")
    }
  }
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.runtime

import spray.json.JsValue

import net.ruippeixotog.scalafbp.component.{ Component, InPort, OutPort }

case class Graph(
    id: String,
    nodes: Map[String, Node] = Map.empty,
    publicIn: Map[String, PublicPort] = Map.empty,
    publicOut: Map[String, PublicPort] = Map.empty) {

  def edges: Map[PortRef, Map[PortRef, Edge]] =
    for { (srcNode, node) <- nodes; (srcPort, tgts) <- node.edges }
      yield (PortRef(srcNode, srcPort), tgts)

  def initials: Map[PortRef, Initial] =
    for { (tgtNode, node) <- nodes; (tgtPort, initials) <- node.initials }
      yield (PortRef(tgtNode, tgtPort), initials)

  def inPortInfo(ref: PortRef): Option[InPort[_]] =
    nodes.get(ref.node).flatMap(_.component.inPorts.find(_.id == ref.port))

  def outPortInfo(ref: PortRef): Option[OutPort[_]] =
    nodes.get(ref.node).flatMap(_.component.outPorts.find(_.id == ref.port))
}

case class Node(
  component: Component,
  metadata: Map[String, JsValue] = Map.empty,
  edges: Map[String, Map[PortRef, Edge]] = Map.empty,
  initials: Map[String, Initial] = Map.empty)

case class PortRef(node: String, port: String) {
  override def toString = s"$node[$port]"
}

case class Edge(metadata: Map[String, JsValue] = Map.empty)
case class Initial(value: JsValue, metadata: Map[String, JsValue] = Map.empty)
case class PublicPort(internal: PortRef, metadata: Map[String, JsValue] = Map.empty) 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component

import akka.actor.{ Actor, Props }
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component.ComponentActor.Incoming

case object ErrorComponent extends Component {
  val name = "ErrorComponent"
  val description = ""
  val icon = None

  val inPorts = List(InPort[JsValue](s"in", ""))
  val outPorts = Nil

  val instanceProps = Props(new Actor {
    def receive = {
      case Incoming(_, _) => throw new RuntimeException("process error")
    }
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.core

import akka.actor.Props
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._
import net.ruippeixotog.scalafbp.util.NashornEngine

case object MakeFunction extends Component {
  val name = "core/MakeFunction"
  val description = "Evaluates a JavaScript function each time data hits an input port " +
    "and sends the return value to the output port"

  val icon = Some("forward")

  val inPort = InPort[JsValue]("in", "Packet to be processed")
  val funcPort = InPort[String]("func", "Function to evaluate. The variable 'x' refers to the input; " +
    "for example, 'return x * 2' doubles the value of the input packet.'")
  val inPorts = List(inPort, funcPort)

  val outPort = OutPort[JsValue]("out", "Forwarded packet")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) with NashornEngine {
    val func = funcPort.stream.map(JsFunction(_))
    inPort.stream.withLatestFrom(func) { (x, f) => f(x) }.pipeTo(outPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.core

import akka.actor.Props
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._

case object Kick extends Component {
  val name = "core/Kick"
  val description = "Generates a packet everytime a signal is received"
  val icon = Some("share")

  val inPort = InPort[JsValue]("in", "Packet to be sent")
  val kickPort = InPort[Unit]("kick", "Signal to send the data packet")
  val inPorts = List(inPort, kickPort)

  val outPort = OutPort[JsValue]("out", "The kicked packet")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) {
    kickPort.stream.withLatestFrom(inPort.stream) { (_, in) => in }.pipeTo(outPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.core

import akka.actor.Props
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._

case object Repeat extends Component {
  val name = "core/Repeat"
  val description = "Forwards packets in the same way it receives them"
  val icon = Some("forward")

  val inPorts = List(
    InPort[JsValue]("in", "Packet to forward"))

  val outPorts = List(
    OutPort[JsValue]("out", "Forwarded packet"))

  val instanceProps = Props(new ComponentActor(this) {
    inPorts.head.stream.pipeTo(outPorts.head)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.core

import akka.actor.Props
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component.ComponentActor.OnAllInputPortsClosed
import net.ruippeixotog.scalafbp.component._

case object Output extends Component {
  val name = "core/Output"
  val description = "Sends packets to clients as output messages"
  val icon = Some("bug")

  val inPorts = List(
    InPort[JsValue]("in", "Packet to be sent as output message"))

  val outPorts = List(
    OutPort[JsValue]("out", "The sent packets"))

  val instanceProps = Props(new ComponentActor(this) {
    override val terminationPolicy = List(OnAllInputPortsClosed)

    inPorts.head.stream
      .doOnEach { data => broker ! ComponentActor.Message(data.compactPrint) }
      .pipeTo(outPorts.head)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.core

import scala.concurrent.duration._

import akka.actor.Props
import rx.lang.scala.Observable
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._

case object RepeatDelayed extends Component {
  val name = "core/RepeatDelayed"
  val description = "Forwards packets after a set delay"
  val icon = Some("clock-o")

  val inPort = InPort[JsValue]("in", "Packet to forward with a delay")
  val delayPort = InPort[Long]("delay", "Delay length (ms)")
  val inPorts = List(inPort, delayPort)

  val outPort = OutPort[JsValue]("out", "Forwarded packet")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) {
    val str = inPort.stream
      .withLatestFrom(delayPort.stream)((_, _))
      .flatMap { case (in, delay) => Observable.just(in).delay(delay.millis) }
      .pipeTo(outPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.stream

import akka.actor.Props
import spray.json.DefaultJsonProtocol._
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._

case object ToArray extends Component {
  val name = "stream/ToArray"
  val description = "Consumes all the packets of a stream to emit a single array"
  val icon = None

  val inPort = InPort[JsValue]("in", "The input stream")
  val inPorts = List(inPort)

  val arrayPort = OutPort[List[JsValue]]("array", "The input stream packed as an array")
  val outPorts = List(arrayPort)

  val instanceProps = Props(new ComponentActor(this) {
    inPort.stream.toList.pipeTo(arrayPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.stream

import akka.actor.Props
import rx.lang.scala.Observable
import spray.json.{ JsArray, JsValue }

import net.ruippeixotog.scalafbp.component._
import net.ruippeixotog.scalafbp.util.NashornEngine

case object MapConcat extends Component {
  val name = "stream/MapConcat"
  val description = "Transforms the elements of a stream into arrays of elements and flatterns them"
  val icon = Some("code")

  val inPort = InPort[JsValue]("in", "The stream to transform")
  val funcPort = InPort[String]("func", "The function with argument x to use for transformation. " +
    "Must return an array. While not defined, all elements pass untouched.")
  val inPorts = List(inPort, funcPort)

  val outPort = OutPort[JsValue]("out", "The transformed stream")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) with NashornEngine {
    val defaultFunc = Observable.just[JsFunction](JsArray(_))

    val func = defaultFunc ++ funcPort.stream.map(JsFunction(_))
    inPort.stream.withLatestFrom(func) { (x, f) => f(x) }.flatMapIterable {
      case JsArray(elems) => elems
      case js => throw new IllegalArgumentException(
        s"The value ${js.compactPrint} returned by the function is not an array")
    }.pipeTo(outPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.stream

import akka.actor.Props
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._
import net.ruippeixotog.scalafbp.util.NashornEngine

case object Scan extends Component {
  val name = "stream/Scan"
  val description = "Accumulates the elements of a stream using a function, emitting the intermediate values"
  val icon = Some("compress")

  val inPort = InPort[JsValue]("in", "The stream to scan")
  val initialPort = InPort[JsValue]("initial", "The initial element")
  val funcPort = InPort[String]("func", "A function with arguments (acc, x) used to create the next element")
  val inPorts = List(inPort, initialPort, funcPort)

  val outPort = OutPort[JsValue]("out", "The generated elements")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) with NashornEngine {
    val in = inPort.stream
    val initial = initialPort.stream.head
    val func = funcPort.stream.head.map(JsFunction2(_, "acc", "x"))

    initial.zip(func).flatMap { case (init, f) => in.scan(init)(f) }.pipeTo(outPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.stream

import akka.actor.Props
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._

case object Take extends Component {
  val name = "stream/Take"
  val description = "Emits only the first N elements of a stream"
  val icon = Some("filter")

  val inPort = InPort[JsValue]("in", "The stream to take elements from")
  val nPort = InPort[Int]("n", "The number of elements to take")
  val inPorts = List(inPort, nPort)

  val outPort = OutPort[JsValue]("out", "The taken elements")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) {
    val in = inPort.stream
    val toTake = nPort.stream.head

    toTake.flatMap(in.take).doOnCompleted(context.stop(self)).pipeTo(outPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.stream

import akka.actor.Props
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._
import net.ruippeixotog.scalafbp.util.NashornEngine

case object ZipWith extends Component {
  val name = "stream/ZipWith"
  val description = "Combines elements from two streams in pairs using a selector function"
  val icon = Some("dropbox")

  val in1Port = InPort[JsValue]("in1", "The first stream")
  val in2Port = InPort[JsValue]("in2", "The second stream")
  val selectorPort = InPort[String]("selector", "A function with arguments (x1, x2) used to create the zipped element")
  val inPorts = List(in1Port, in2Port, selectorPort)

  val outPort = OutPort[JsValue]("out", "The zipped stream")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) with NashornEngine {
    val selector = selectorPort.stream.head.map(JsFunction2(_, "x1", "x2"))

    selector.flatMap(in1Port.stream.zipWith(in2Port.stream))
      .doOnCompleted(context.stop(self))
      .pipeTo(outPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.stream

import akka.actor.Props
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._
import net.ruippeixotog.scalafbp.util.NashornEngine

case object Map extends Component {
  val name = "stream/Map"
  val description = "Transforms the elements of a stream using a function"

  val icon = Some("code")

  val inPort = InPort[JsValue]("in", "The stream to transform")
  val funcPort = InPort[String]("func", "The function with argument x to use for transformation. " +
    "While not defined, all elements pass unmodified.")
  val inPorts = List(inPort, funcPort)

  val outPort = OutPort[JsValue]("out", "The transformed stream")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) with NashornEngine {
    val defaultFunc: JsFunction = identity
    val func = defaultFunc +: funcPort.stream.map(JsFunction(_))
    inPort.stream.withLatestFrom(func) { (x, f) => f(x) }.pipeTo(outPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.stream

import akka.actor.Props
import spray.json.DefaultJsonProtocol._
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._

case object FromArray extends Component {
  val name = "stream/FromArray"
  val description = "Receives a stream of arrays and emits each of their elements"
  val icon = None

  val arrayPort = InPort[List[JsValue]]("array", "The input stream of arrays")
  val inPorts = List(arrayPort)

  val outPort = OutPort[JsValue]("out", "The stream of unpacked data")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) {
    arrayPort.stream.flatMapIterable(identity).pipeTo(outPort)
  })
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.component.stream

import akka.actor.Props
import spray.json.JsValue

import net.ruippeixotog.scalafbp.component._

case object Drop extends Component {
  val name = "stream/Drop"
  val description = "Drops the first N elements of a stream and emits the remaining ones"
  val icon = Some("filter")

  val inPort = InPort[JsValue]("in", "The stream to drop elements from")
  val nPort = InPort[Int]("n", "The number of elements to drop")
  val inPorts = List(inPort, nPort)

  val outPort = OutPort[JsValue]("out", "The elements that were not dropped")
  val outPorts = List(outPort)

  val instanceProps = Props(new ComponentActor(this) {
    val in = inPort.stream
    val toDrop = nPort.stream.head

    toDrop.flatMap(in.drop).pipeTo(outPort)
  })
}