<commit_before>package core

import akka.actor.ActorRef
import akka.http.scaladsl.model.ws.{Message, TextMessage}
import akka.stream.scaladsl.{Flow, GraphDSL, Merge, Sink, Source}
import akka.stream.{FlowShape, Materializer, OverflowStrategy}
import akka.util.ByteString
import core.CoreMessage._

import scala.concurrent.ExecutionContext.Implicits.global

class Websocket(val provider : ActorRef, val port : Int, implicit val mat : Materializer) {
  val playerActorSource= Source.actorRef[Any](10000,OverflowStrategy.fail)

  def flow(id: String) : Flow[Message,Message, Any] = Flow.fromGraph(GraphDSL.create(playerActorSource) { implicit builder => playerActor =>
    import GraphDSL.Implicits._
  //  println("Websocket flow called " + id + " on port : " + port)
    val materialization = builder.materializedValue.map(playerActorRef => AddClient(id, playerActorRef))
    val merge = builder.add(Merge[Any](2))

    val messageToEventFlow = builder.add(Flow[Any].map {
      case TextMessage.Strict(txt) => ClientInputWithLocation(id,txt)
      case x:TextMessage.Streamed => {
        val sink = Sink.fold[String,String]("")(_+_)
        val truc = x.getStreamedText.runWith(Sink.fold("")(_+_),mat)
          truc.onComplete( t =>   provider !  ClientInputWithLocation(id,t.get ) )
      }
    })
    val eventToMessageFlow= builder.add(Flow[Any].map{
      case PlayersUpdate(json) =>TextMessage(json)
      case PlayersUpdateRaw(bytes) =>  akka.http.scaladsl.model.ws.BinaryMessage(ByteString.fromByteBuffer(bytes))
    })
    val ProviderSink = Sink.actorRef[Any](provider,DeleteClient(id))

    materialization ~> merge
    messageToEventFlow ~> merge
    merge ~> ProviderSink
    playerActor ~>  eventToMessageFlow
    FlowShape(messageToEventFlow.in, eventToMessageFlow.out)
  })
}


<commit_msg>Add websocket flow schema<commit_after>package core

import akka.actor.ActorRef
import akka.http.scaladsl.model.ws.{Message, TextMessage}
import akka.stream.scaladsl.{Flow, GraphDSL, Merge, Sink, Source}
import akka.stream.{FlowShape, Materializer, OverflowStrategy}
import akka.util.ByteString
import core.CoreMessage._

import scala.concurrent.ExecutionContext.Implicits.global

class Websocket(val provider : ActorRef, val port : Int, implicit val mat : Materializer) {
  val playerActorSource= Source.actorRef[Any](10000,OverflowStrategy.fail)

  def flow(id: String) : Flow[Message,Message, Any] = Flow.fromGraph(GraphDSL.create(playerActorSource) { implicit builder => playerActor =>
    import GraphDSL.Implicits._
  //  println("Websocket flow called " + id + " on port : " + port)
    val materialization = builder.materializedValue.map(playerActorRef => AddClient(id, playerActorRef))
    val merge = builder.add(Merge[Any](2))

    val messageToEventFlow = builder.add(Flow[Any].map {
      case TextMessage.Strict(txt) => ClientInputWithLocation(id,txt)
      case x:TextMessage.Streamed => {
        val sink = Sink.fold[String,String]("")(_+_)
        val truc = x.getStreamedText.runWith(Sink.fold("")(_+_),mat)
          truc.onComplete( t =>   provider !  ClientInputWithLocation(id,t.get ) )
      }
    })
    val eventToMessageFlow= builder.add(Flow[Any].map{
      case PlayersUpdate(json) =>TextMessage(json)
      case PlayersUpdateRaw(bytes) =>  akka.http.scaladsl.model.ws.BinaryMessage(ByteString.fromByteBuffer(bytes))
    })
    val ProviderSink = Sink.actorRef[Any](provider,DeleteClient(id))

    materialization ~> merge
    messageToEventFlow ~> merge
    merge ~> ProviderSink
    playerActor ~>  eventToMessageFlow
    FlowShape(messageToEventFlow.in, eventToMessageFlow.out)
  })
}
/*


      Client                    Command
        +                          +
        |                          |
        |                          |
        v                          |
+-------+-------+       +----------v---------+
|               |       |                    |
|materialization|       | messageToEventFlow |
|               |       |                    |
+-------+-------+       +----------+---------+
        |                          |
        |                          |
        +--------+   +-------------+
                 |   |
                 |   |
                 |   |
                 |   |
                 |   |
                 |   |
               +-v---v--+
               |        |
               | merge  |
               |        |                     Client
               +---+----+                       +
                   |                            |
                   |                            |
                   |                 +----------v---------+
           +-------v-------+         |                    |
           |               |         | eventToMessageFlow |
           |  ProviderSink |         |                    |
           |               |         +--------------------+
           +---------------+




 */


{'fraction_non_alphanumeric': 0.1404471544715447, 'fraction_numerical': 0.0024390243902439024, 'mean_word_length': 1.780225988700565, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 11, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4431828', 'n_tokens_mistral': 1406, 'n_tokens_neox': 1368, 'n_words': 222}