vasily-kirichenko

// download https://drive.google.com/file/d/1kXYcIkDMX25km5YtSBfFa9Xxx2dx8cNB/view?usp=sharing locally

using System;
using System.IO;
using System.Security.Cryptography.X509Certificates;

namespace EccSigNet5
{
    static class Program
    {

vasily-kirichenko

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import java.io.File
import java.io.IOException
import kotlin.coroutines.resumeWithException

@ExperimentalCoroutinesApi
suspend fun startProcess(cmd: String, workingDir: File): Process = suspendCancellableCoroutine { cont ->
    try {
        val process =

vasily-kirichenko

forAll { (i: Array[Char]) => {
  val sb = new StringBuilder
  for (c <- i) {
    c match {
      case '\\' => sb += '\\' += '\\'
      case '"' => sb += '\\' += '"'
      case _ => sb += c
    }
  }
  val expected = sb.toString

vasily-kirichenko

2019-01-25 11:04:29
Full thread dump OpenJDK 64-Bit Server VM (25.152-b26 mixed mode):

"ApplicationImpl pooled thread 763" #2017 daemon prio=4 os_prio=-1 tid=0x000000004b056800 nid=0x4d90 waiting on condition [0x0000000070bcf000]
   java.lang.Thread.State: TIMED_WAITING (parking)
	at sun.misc.Unsafe.park(Native Method)
	- parking to wait for  <0x00000000aa188de0> (a java.util.concurrent.SynchronousQueue$TransferStack)
	at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	at java.util.concurrent.SynchronousQueue$TransferStack.awaitFulfill(SynchronousQueue.java:460)
	at java.util.concurrent.SynchronousQueue$TransferStack.transfer(SynchronousQueue.java:362)

vasily-kirichenko

2018-12-22 09:28:32
Full thread dump OpenJDK 64-Bit Server VM (25.152-b26 mixed mode):

"Attach Listener" #558 daemon prio=9 os_prio=31 tid=0x00007f98d6b16800 nid=0x12f47 waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

   Locked ownable synchronizers:
	- None

"Timer-0" #26 daemon prio=5 os_prio=31 tid=0x00007f98cfd2c800 nid=0xdf17 in Object.wait() [0x00007000082ab000]

vasily-kirichenko

import akka.NotUsed
import akka.actor.Scheduler
import akka.actor.typed.{ActorRef, _}
import akka.actor.typed.scaladsl.AskPattern._
import akka.actor.typed.scaladsl.Behaviors
import akka.actor.typed.scaladsl.adapter._
import akka.cluster.Cluster
import akka.cluster.ddata.{LWWMap, LWWMapKey, ReplicatedData}
import akka.cluster.ddata.typed.scaladsl.{Replicator, _}
import akka.stream._

vasily-kirichenko

def cache[In, Out](flow: Flow[In, Out, _]) = Flow.fromGraph(GraphDSL.create() { implicit b =>
    import GraphDSL.Implicits._

    var cache: Map[In, Out] = Map.empty
    val partition = b.add(Partition[In](2, in => if (cache.contains(in)) 0 else 1))
    val merge = b.add(Merge[Out](2))
    partition.out(0).map(in => cache(in)) ~> merge

    val broadcast = b.add(Broadcast[In](2))
    partition.out(1) ~> broadcast

vasily-kirichenko

def processor(account: String, taskCh: MVar[Int]): Observable[String] =
    Observable.fromAsyncStateAction[Unit, String](_ =>
        for {
            task <- taskCh.take
            // simulate real processing by an async sleep
            _ <- Task.sleep(Random.nextInt(500).millis)
        } yield {
            println(s"[$account] processed $task")
            ((task + 100).toString, ())
        }

vasily-kirichenko

open Akka.Actor
open Akka.Streams
open Akka.Streams.Dsl
open System
open System.Threading

[<EntryPoint>]
let main _ =
    let sys = ActorSystem.Create "test"
    let mat = ActorMaterializer.Create sys

vasily-kirichenko

let props (_scanType: ScanType) : Akka.Actor.Props =
    props(
        let rec start (ctx: Actor<obj>) (state: State) (msg: obj) = 
            match msg with
            | LifecycleEvent PostStop -> kill state.Process
            | :? Msg as msg ->
                match msg with
                | Start -> 
                    Starter.start def |> Async.map Started |!> ctx.Self.Retype()
                    become (starting ctx state)