<commit_before>package org.http4s.grizzly

import play.api.libs.iteratee.{Step, Done, Input, Iteratee}
import org.http4s._
import org.glassfish.grizzly.WriteHandler
import concurrent.{ExecutionContext, Future, Promise}
import org.glassfish.grizzly.http.server.io.NIOOutputStream

/**
 * @author <NAME>
 * Created on 2/11/13 at 8:44 AM
 */
class OutputIteratee(os: NIOOutputStream, chunkSize: Int)(implicit executionContext: ExecutionContext) extends Iteratee[HttpChunk,Unit] {

  private[this] var osFuture: Future[Unit] = Future.successful()

  private[this] def writeBytes(bytes: Raw): Unit = {
    val promise: Promise[Unit] = Promise()

    val asyncWriter = new  WriteHandler {
      override def onError(t: Throwable) {
        promise.failure(t)
        sys.error(s"Error on write listener: ${t.getStackTraceString}")
      }
      override def onWritePossible() = promise.success(os.write(bytes))
    }
    // Synchronized so that our osFuture doesn't get ruined by crazy enumerators that may call
    // push from different threads or something crazy like that.
    synchronized(osFuture = osFuture.flatMap{ _ => os.notifyCanWrite(asyncWriter,bytes.length); promise.future })
  }

  // Create a buffer for storing data until its larger than the chunkSize
  private[this] val buff = new Array[Byte](chunkSize)
  private[this] var buffSize = 0

  private[this] def push(in: Input[HttpChunk]): Iteratee[HttpChunk,Unit] = {
    in match {
      case Input.Empty => this
      case Input.EOF =>
        if (buffSize > 0) {
          writeBytes(buff.take(buffSize))
          buffSize = 0
        }
        Iteratee.flatten(osFuture.map(Done(_)))

      case Input.El(chunk) => chunk match {
        case HttpEntity(bytes) =>
          // Do the buffering
          if (bytes.length + buffSize >= chunkSize) {
            val tmp = new Array[Byte](bytes.length + buffSize)
            buff.take(buffSize).copyToArray(tmp)
            bytes.copyToArray(tmp,buffSize)
            writeBytes(tmp)
            buffSize = 0
          } else {
            bytes.copyToArray(buff, buffSize)
            buffSize += bytes.length
          }

        case _ => sys.error("Griz output Iteratee doesn't support your data type!")
      }

      this
    }
  }<commit_msg>Moved the synchronized to make the whole continuation of grizzly OutputIteratee safe if called by non sequential Enumerators.<commit_after>package org.http4s.grizzly

import play.api.libs.iteratee.{Step, Done, Input, Iteratee}
import org.http4s._
import org.glassfish.grizzly.WriteHandler
import concurrent.{ExecutionContext, Future, Promise}
import org.glassfish.grizzly.http.server.io.NIOOutputStream

/**
 * @author <NAME>
 * Created on 2/11/13 at 8:44 AM
 */
class OutputIteratee(os: NIOOutputStream, chunkSize: Int)(implicit executionContext: ExecutionContext) extends Iteratee[HttpChunk,Unit] {

  private[this] var osFuture: Future[Unit] = Future.successful()

  private[this] def writeBytes(bytes: Raw): Unit = {
    val promise: Promise[Unit] = Promise()

    val asyncWriter = new  WriteHandler {
      override def onError(t: Throwable) {
        promise.failure(t)
        sys.error(s"Error on write listener: ${t.getStackTraceString}")
      }
      override def onWritePossible() = promise.success(os.write(bytes))
    }

    osFuture = osFuture.flatMap{ _ => os.notifyCanWrite(asyncWriter,bytes.length); promise.future }
  }

  // Create a buffer for storing data until its larger than the chunkSize
  private[this] val buff = new Array[Byte](chunkSize)
  private[this] var buffSize = 0

  // synchronized so that enumerators that work in different threads cant totally mess it up.
  private[this] def push(in: Input[HttpChunk]): Iteratee[HttpChunk,Unit] = synchronized {
    in match {
      case Input.Empty => this
      case Input.EOF =>
        if (buffSize > 0) {
          writeBytes(buff.take(buffSize))
          buffSize = 0
        }
        Iteratee.flatten(osFuture.map(Done(_)))

      case Input.El(chunk) => chunk match {
        case HttpEntity(bytes) =>
          // Do the buffering
          if (bytes.length + buffSize >= chunkSize) {
            val tmp = new Array[Byte](bytes.length + buffSize)
            buff.take(buffSize).copyToArray(tmp)
            bytes.copyToArray(tmp,buffSize)
            writeBytes(tmp)
            buffSize = 0
          } else {
            bytes.copyToArray(buff, buffSize)
            buffSize += bytes.length
          }

        case _ => sys.error("Griz output Iteratee doesn't support your data type!")
      }

      this
    }
  }
{'fraction_non_alphanumeric': 0.1050328227571116, 'fraction_numerical': 0.00612691466083151, 'mean_word_length': 2.877014418999152, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3954950', 'n_tokens_mistral': 1385, 'n_tokens_neox': 1347, 'n_words': 351}