<commit_before>/**
  * Copyright 2015 <NAME>
  *
  * Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
  * an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  *
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */


package cmwell.ws

import akka.NotUsed
import akka.stream.SourceShape
import akka.stream.scaladsl.{Flow, GraphDSL, Merge, Source}
import akka.util.ByteString
import cmwell.domain.{Infoton, _}
import cmwell.formats.Formatter
import cmwell.fts._
import cmwell.util.{BoxedFailure, EmptyBox, FullBox}
import cmwell.util.concurrent.retry
import cmwell.util.string._
import cmwell.util.os.Props.os.getAvailableProcessors
import cmwell.util.stream.TakeWeighted
import cmwell.ws.util.DateParser
import com.typesafe.scalalogging.LazyLogging
import logic.CRUDServiceFS
import org.joda.time.DateTime

import scala.concurrent.{ExecutionContext, Future, duration}
import duration.DurationInt
import scala.util.{Success, Try}

object Streams extends LazyLogging {

  //CONSTS
  private val fsp = FieldSortParams(List("system.indexTime" -> Asc))
  val bo0 = scala.collection.breakOut[Seq[Infoton], ByteString, List[ByteString]]
  val bo1 = scala.collection.breakOut[Seq[Infoton], String, List[String]]
  val bo2 = scala.collection.breakOut[Seq[SearchThinResult], String, List[String]]
  val bo3 = scala.collection.breakOut[Seq[Infoton], SearchThinResult, List[SearchThinResult]]
  val parallelism = math.min(cmwell.util.os.Props.os.getAvailableProcessors,Settings.cassandraBulkSize)
  val endln = ByteString(cmwell.util.os.Props.endln)

  object Flows {

    def infotonToByteString(formatter: Formatter): Flow[Infoton,ByteString,NotUsed] = Flow.fromFunction[Infoton,ByteString] { infoton =>
      val formatted = formatter.render(infoton)
      val trimmed = dropRightWhile(formatted)(_.isWhitespace)
      ByteString(trimmed)
    }.intersperse(endln)

    def searchThinResultToByteString(formatter: Formatter): Flow[SearchThinResult,ByteString,NotUsed] = Flow.fromFunction[SearchThinResult,ByteString] { str =>
      val formatted = formatter.render(str)
      val trimmed = dropRightWhile(formatted)(_.isWhitespace)
      ByteString(trimmed)
    }.intersperse(endln)

    def formattableToByteString(formatter: Formatter, length: Option[Long] = None): Flow[Formattable,ByteString,NotUsed] = {
      if(formatter.mimetype.contains("json")) {
        val s = Flow[Formattable].mapConcat {
          case IterationResults(_, _, Some(iSeq), _, _) => iSeq.toList
          case InfotonHistoryVersions(iSeq) => iSeq.toList
          case BagOfInfotons(iSeq) => iSeq.toList
          case i: Infoton => List(i)
//          case str: SearchThinResult => List(str)
//          case str: SearchThinResults => str.thinResults.toList
        }

        length.fold(s)(l => s.take(l)).map { i =>
          val rv = Try(formatter.render(i))
          rv.failed.foreach { err =>
            logger.error(s"error during formating of $i", err)
          }
          rv
        }.collect {
          case Success(renderedString) =>
            if (renderedString.endsWith(cmwell.util.os.Props.endln)) ByteString(renderedString)
            else ByteString(renderedString + cmwell.util.os.Props.endln)
        }
      }
      else {
        val f2bs: Formattable => ByteString = { formattable =>
          val formatted = formatter.render(formattable)
          val trimmed = dropRightWhile(formatted)(_.isWhitespace)
          ByteString(trimmed)
        }

        length.fold(Flow.fromFunction(f2bs))(l => Flow.fromGraph(TakeWeighted[Formattable](l,true,{
          case IterationResults(_, _, iSeqOpt, _, _) => iSeqOpt.fold(0)(_.size).toLong
          case InfotonHistoryVersions(iSeq) => iSeq.size.toLong
          case BagOfInfotons(iSeq) => iSeq.size.toLong
          case i: Infoton => 1L
        })).map(f2bs)).intersperse(endln)
      }
    }

    val searchThinResultToFatInfoton: Flow[SearchThinResult,Infoton,NotUsed] = Flow[SearchThinResult]
      .mapAsyncUnordered(parallelism)(str => CRUDServiceFS.getInfotonByUuidAsync(str.uuid))
      .collect{ case FullBox(i) => i}

    val searchThinResultsToFatInfotons: Flow[SearchThinResults,Infoton,NotUsed] = Flow[SearchThinResults]
      .mapConcat { case SearchThinResults(_, _, _, str, _) => str.toList }
      .via(searchThinResultToFatInfoton)

    val iterationResultsToFatInfotons: Flow[IterationResults,Infoton,NotUsed] = Flow[IterationResults]
      .collect { case IterationResults(_, _, Some(iSeq), _, _) => iSeq}
      .mapConcat(_.map(_.uuid)(bo1))
      .mapAsyncUnordered(parallelism)(CRUDServiceFS.getInfotonByUuidAsync)
      .collect{ case FullBox(i) => i}

    val iterationResultsToInfotons: Flow[IterationResults,Infoton,NotUsed] = Flow[IterationResults]
      .collect { case IterationResults(_, _, Some(iSeq), _, _) => iSeq}<commit_msg>fixed a bug causing stream to not end with a newline (#21)<commit_after>/**
  * Copyright 2015 <NAME>
  *
  * Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
  * an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  *
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */


package cmwell.ws

import akka.NotUsed
import akka.stream.SourceShape
import akka.stream.scaladsl.{Flow, GraphDSL, Merge, Source}
import akka.util.ByteString
import cmwell.domain.{Infoton, _}
import cmwell.formats.Formatter
import cmwell.fts._
import cmwell.util.{BoxedFailure, EmptyBox, FullBox}
import cmwell.util.concurrent.retry
import cmwell.util.string._
import cmwell.util.os.Props.os.getAvailableProcessors
import cmwell.util.stream.TakeWeighted
import cmwell.ws.util.DateParser
import com.typesafe.scalalogging.LazyLogging
import logic.CRUDServiceFS
import org.joda.time.DateTime

import scala.concurrent.{ExecutionContext, Future, duration}
import duration.DurationInt
import scala.util.{Success, Try}

object Streams extends LazyLogging {

  //CONSTS
  private val fsp = FieldSortParams(List("system.indexTime" -> Asc))
  val bo0 = scala.collection.breakOut[Seq[Infoton], ByteString, List[ByteString]]
  val bo1 = scala.collection.breakOut[Seq[Infoton], String, List[String]]
  val bo2 = scala.collection.breakOut[Seq[SearchThinResult], String, List[String]]
  val bo3 = scala.collection.breakOut[Seq[Infoton], SearchThinResult, List[SearchThinResult]]
  val parallelism = math.min(cmwell.util.os.Props.os.getAvailableProcessors,Settings.cassandraBulkSize)
  val endln = ByteString(cmwell.util.os.Props.endln)

  object Flows {

    def infotonToByteString(formatter: Formatter): Flow[Infoton,ByteString,NotUsed] = Flow.fromFunction[Infoton,ByteString] { infoton =>
      val formatted = formatter.render(infoton)
      val trimmed = dropRightWhile(formatted)(_.isWhitespace)
      ByteString(trimmed) ++ endln
    }

    def searchThinResultToByteString(formatter: Formatter): Flow[SearchThinResult,ByteString,NotUsed] = Flow.fromFunction[SearchThinResult,ByteString] { str =>
      val formatted = formatter.render(str)
      val trimmed = dropRightWhile(formatted)(_.isWhitespace)
      ByteString(trimmed) ++ endln
    }

    def formattableToByteString(formatter: Formatter, length: Option[Long] = None): Flow[Formattable,ByteString,NotUsed] = {
      if(formatter.mimetype.contains("json")) {
        val s = Flow[Formattable].mapConcat {
          case IterationResults(_, _, Some(iSeq), _, _) => iSeq.toList
          case InfotonHistoryVersions(iSeq) => iSeq.toList
          case BagOfInfotons(iSeq) => iSeq.toList
          case i: Infoton => List(i)
//          case str: SearchThinResult => List(str)
//          case str: SearchThinResults => str.thinResults.toList
        }

        length.fold(s)(l => s.take(l)).map { i =>
          val rv = Try(formatter.render(i))
          rv.failed.foreach { err =>
            logger.error(s"error during formating of $i", err)
          }
          rv
        }.collect {
          case Success(renderedString) =>
            if (renderedString.endsWith(cmwell.util.os.Props.endln)) ByteString(renderedString)
            else ByteString(renderedString + cmwell.util.os.Props.endln)
        }
      }
      else {
        val f2bs: Formattable => ByteString = { formattable =>
          val formatted = formatter.render(formattable)
          val trimmed = dropRightWhile(formatted)(_.isWhitespace)
          ByteString(trimmed + cmwell.util.os.Props.endln)
        }

        length.fold(Flow.fromFunction(f2bs))(l => Flow.fromGraph(TakeWeighted[Formattable](l,true,{
          case IterationResults(_, _, iSeqOpt, _, _) => iSeqOpt.fold(0)(_.size).toLong
          case InfotonHistoryVersions(iSeq) => iSeq.size.toLong
          case BagOfInfotons(iSeq) => iSeq.size.toLong
          case i: Infoton => 1L
        })).map(f2bs))
      }
    }

    val searchThinResultToFatInfoton: Flow[SearchThinResult,Infoton,NotUsed] = Flow[SearchThinResult]
      .mapAsyncUnordered(parallelism)(str => CRUDServiceFS.getInfotonByUuidAsync(str.uuid))
      .collect{ case FullBox(i) => i}

    val searchThinResultsToFatInfotons: Flow[SearchThinResults,Infoton,NotUsed] = Flow[SearchThinResults]
      .mapConcat { case SearchThinResults(_, _, _, str, _) => str.toList }
      .via(searchThinResultToFatInfoton)

    val iterationResultsToFatInfotons: Flow[IterationResults,Infoton,NotUsed] = Flow[IterationResults]
      .collect { case IterationResults(_, _, Some(iSeq), _, _) => iSeq}
      .mapConcat(_.map(_.uuid)(bo1))
      .mapAsyncUnordered(parallelism)(CRUDServiceFS.getInfotonByUuidAsync)
      .collect{ case FullBox(i) => i}

    val iterationResultsToInfotons: Flow[IterationResults,Infoton,NotUsed] = Flow[IterationResults]
      .collect { case IterationResults(_, _, Some(iSeq), _, _) => iSeq}
{'fraction_non_alphanumeric': 0.11019626615605553, 'fraction_numerical': 0.0036381043561512687, 'mean_word_length': 3.7851580393953275, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2969711', 'n_tokens_mistral': 3314, 'n_tokens_neox': 3236, 'n_words': 665}