6

我正在尝试制作一个将流写入分桶文件的接收器：当达到特定条件（时间、文件大小等）时，当前输出流将关闭，新输出流将打开到新的桶文件。

我检查了如何在io对象中创建不同的接收器，但没有太多示例。因此，我尝试遵循如何编写resource和chunkW编写。我最终得到了以下代码，为简单起见，桶现在只用一个表示Int，但最终会是某种类型的输出流。

  val buckets: Channel[Task, String, Int] = {

    //recursion to step through the stream
    def go(step: Task[String => Task[Int]]): Process[Task, String => Task[Int]] = {

      // Emit the value and repeat
      def next(msg: String => Task[Int]) =
        Process.emit(msg) ++
          go(step)


      Process.await[Task, String => Task[Int], String => Task[Int]](step)(
        next
        , Process.halt // TODO ???
        , Process.halt) // TODO ???
    }

   //starting bucket
    val acquire: Task[Int] = Task.delay {
      val startBuck = nextBucket(0)
      println(s"opening bucket $startBuck")
      startBuck
    }

   //the write step
    def step(os: Int): Task[String => Task[Int]] =
      Task.now((msg: String) => Task.delay {
        write(os, msg)
        val newBuck = nextBucket(os)
        if (newBuck != os) {
          println(s"closing bucket $os")
          println(s"opening bucket $newBuck")
        }
        newBuck
      })

    //start the Channel
    Process.await(acquire)(
      buck => go(step(buck))
      , Process.halt, Process.halt)
  }

 def write(bucket: Int, msg: String) { println(s"$bucket\t$msg") }
 def nextBucket(b: Int) = b+1

这其中有很多问题：

  1. step在开始时将桶传递一次，并且在递归期间永远不会改变。我不确定如何在递归go中创建一个step将使用前一个任务中的存储桶（Int）的新任务，因为我必须提供一个字符串才能完成该任务。
  2. 调用的fallbackandcleanup没有await收到结果rcv（如果有的话）。在io.resource函数中，由于资源是固定的，它可以正常工作，但是，在我的情况下，资源可能会在任何步骤发生变化。我将如何将对当前打开的存储桶的引用传递给这些回调？
4

1 回答 1

0

那么其中一个选项（即时间）可能是go在水槽上使用简单。这个使用基于时间的，基本上每隔一小时重新打开文件：

val metronome =  Process.awakeEvery(1.hour).map(true)


def writeFileSink(file:String):Sink[Task,ByteVector] = ???


def timeBasedSink(prefix:String) = {
  def go(index:Int) : Sink[Task,ByteVector] = {
    metronome.wye(write(prefix + "_" + index))(wye.interrupt) ++ go(index + 1)
  }

  go(0)
}

对于其他选项（即写入的字节），您可以使用类似的技术，只需保持写入的字节信号并将其与接收器结合。

于 2015-08-29T04:35:02.603 回答