uuid-producer.scala

import scalaz.stream.nio.{file => nio}
import scodec.bits.ByteVector

...

  val uuidSource: Process[Task, String] = 
    Process.repeatEval(
      Task.delay(
        // uuid creation
      )
    )
 
  
  def kafkaSink(producerConfig: ProducerConfig, topic: String): Sink[Task, String] = {
    val producer = new Producer[String, String](producerConfig)
    io.channel { msg =>
      Task.delay {
        val data = new KeyedMessage[String, String](topic, msg, msg)
        producer.send(data)
      }
    }
  }
  
  val path = "D:/"
  val eol = ByteVector(Array(13,10).map(_.toByte))
  val fileSink: Sink[Task, Vector[String]] = io.channel { lines =>
    Task.delay {
       val fname = lines.head concat ".txt"
       val bytes = lines.map(line => ByteVector(line.getBytes))
       Process.emitAll(bytes).toSource.intersperse(eol).to(nio.chunkW(path concat fname)).run.run
    }
  }
 
  def main(args: Array[String]) {
  
    // config
    
        implicit val scheduler = DefaultScheduler

	val proc = 
	 Process.awakeEvery(500 milliseconds)
        .zipWith(uuidSource)((_, uuid) => uuid)
	.observe(kafkaSink(config, topic))
	.chunk(1000)
	.to(fileSink)
		
	proc.run.run
  
  }
  
 ...