Blogs
200 Reasons Why You Shouldn’t Go Down to the Basement
BY Joe
1 hour ago • Read More
Underneath the product aisles of a well-known Monoprix supermarket branch in central Paris, where I've often popped into myself to pick up a bottle of wine and some olives before attending parti...The Forgotten Treehouse Bars of Bygone Summers in Paris
By MessyNessy
1 hour ago • Read More
There was once a place that drew crowds of Parisians away from their grand boulevards and sidewalk cafés to rediscover their inner child, wine & dine in chestnut tree houses and celebrate su...From Paris to Provence in Under an Hour
By MessyNessy
2 days ago • Read More
You can probably guess from that crazy header at the top of my site that I have a thing for vintage photo collages, and you know I have a thing for miniatures because there's an entire category ...5 Instagram-Ready Restaurants in Paris
By MessyNessy
3 days ago • Read More
Love it or hate, instagram gives everyone a little extra hobby in life that doesn't really do any harm and actually can be seriously useful for travel tips. Keeping up with the local off-beat gems i...The Secret Girls Lunch in Paris
By MessyNessy
4 days ago • Read More
Tell your girlfriends to meet you at the prettiest flower shop in Paris– but only the ones you trust with your innermost secrets. We've found ourselves a hidden new lunch spot... Wait for her ...Postcard from the Eiffel Tower, Paris (Russia)
By MessyNessy
4 days ago • Read More
The first thing I learned today, is that there's a small village in Russia called Paris. With a current population of around 1700 people, the rural village was intentionally named after the French...
Recommendations


Find coverage of my work below. For pieces written by me, please see my Publications Page.
2016
Passion for China drives scholars
China Daily USA | January 29
Stanford students awarded inaugural Schwarzman Scholarship
The Stanford Daily | January 13
Stanford students among winners of new scholarship for study in China
Stanford University News | January 12
2015
Convocation: What it means to be part of Stanford
Stanford Report | September 16
Prepared remarks of 'Finding Your Place' by graduate student Jordan Shapiro, BS '15, at Opening Convocation Ceremony on Sept. 15, 2015
Stanford Report | September 15
Stanford to welcome 1,737 freshmen and transfer students on Tuesday
Stanford Report | September 14
Editorial Board: Evaluating the ASSU executives
The Stanford Daily | January 27
2014
For the new kids on the Farm
The Stanford Daily | September 16
Nobel laureates talk about their research, energy availability
The Stanford Daily | May 13
2013
68 candidates declare for ASSU elections
The Stanford Daily | March 8


Symbol Index for "Not Required Intention"
Symbol Index for diagram
"Not Required Intention"
Abandoned Intention (Class)
Agent Accepted Intention (Class)
Agent Rejected Intention (Class)
Agent Selected Intention (Class)
Fulfilled Intention (Class)
Intention State (Class)
Intention Under Consideration (Class)
Intention/State (Generalization Node)
Not Required Intention (Class)
reason for (Association)
Reason Not Required (Class)
Required Intention (Class)


8 Ijebu - Ode Crescent, Agbara Industrial Estate
<EMAIL_ADDRESS>+234-8052797200
KEY PRODUCT ON SALES:-
1.CULINARY / BEVERAGES RANGE OF PRODUCTS
2.ASSORTMILK PRODUCTSED DRINKS RANGE OF PRODUCTS
3.MILK PRODUCTS
4.MACKEREL RANGE OF PRODUCTS
5.CEREALS / OATS PRODUCTS
6.BULK TISSUE PAPERS
7.DETEERGENTS
8.SANITARY TOWELS , Etc…..


{% extends "section_docs.html" %} {% block title %}Software Collections for Users{% endblock %} {% block content %}
To enable support for Software Collections on your system, so that you can enable and build Software Collections, you need to have the scl-utils and scl-utils-build packages installed.
If the scl-utils and scl-utils-build packages are not already installed on your system, you can install them by typing the following at a shell prompt as root:
yum install scl-utils
The scl-utils package provides the scl
tool that lets you enable Software Collections on your system.
The scl
tool is used to enable a Software Collection and to run applications in the Software Collection environment.
General usage of the scl
tool can be described using the following syntax:
scl action software_collection_1 software_collection_2 command
For example, to directly run Perl with the --version
option in a Software Collection named softwarecollection1, execute the following command:
scl enable software_collection_1 'perl --version'
Alternatively, you can create a wrapper script that shortens the commands for running applications in the Software Collection environment.
{% endblock %}


name: main-template layout: true class: left, top --- class: center
# Future vs. Monix Task
## © 2019 Hermann Hueck
https://github.com/hermannhueck/future-vs-monix-task --- # Abstract *scala.concurrent.Future* is familiar to nearly all Scala devs. This presentation first talks about referential transparency and the IO Monad in general. (Monix *Task* is an impl of the IO Monad.) Then it compares *Future* Monix 3.x *Task* with their Pros and Cons. Interop with *Future*: As Scala's *Future* is used in many environments and libraries, we look at the conversion from *Task* to *Future* and - vice versa - from *Future* to *Task*. I will also take a look at *Task* evaluation, cancelation and memoization as well as tail recursive loops and asynchronous boundaries. The presentation will include a comparative discussion on *ExecutionContext* (required for *Future*) and *Scheduler* (required for *Task*, but only to run it). Often recurring on the valuable Monix *Task* doumentation at https://monix.io/docs/3x/eval/task.html the presentation can also be seen as an introduction to Monix *Task*. --- ## Agenda 1. [Referential Transparency](#ReferentialTransparency) 1. [IO Monad (My Impl)](#IOMonad) 1. [Is Future referentially transparent?](#FutureReferentiallyTransparent) 1. [What is Monix Task?](#WhatIsMonixTask) 1. [Task Evaluation](#TaskEvaluation) 1. [Task Cancelation](#TaskCancelation) 1. [Memoization](#Memoization) 1. [Task Builders](#TaskBuilders) 1. [Converting Future to Task](#FutureToTask) 1. [Task as Monad](#TaskAsMonad) 1. [Tail Recursive Loops](#TailRecursiveLoops) 1. [Async Boundaries](#AsyncBoundaries) 1. [Schedulers](#Schedulers) 1. [Error Handling](#ErrorHandling) 1. [Races](#Races) 1. [Delaying a Task](#DelayingTask) 1. [Parallelism (cats.Parallel)](#Parallelism) 1. [TaskApp](#TaskApp) 1. [CompletableFuture](#CompletableFuture) 1. [Resource Management](#ResourceManagement) 1. [Resources](#Resources) --- class: middle, center name: ReferentialTransparency # 1. Referential Transparency
--- class: large-font ## Referential Transparency
An expression is called referentially transparent if it can be replaced with its corresponding value without changing the program's behavior. This requires that the expression is
___pure___
, that is to say the expression value must be the same for the same inputs and its evaluation must have
___no side effects___
. .footnote[ https://en.wikipedia.org/wiki/Referential_transparency ] --- ## Referential Transparency Benefits
- (Equational) Reasoning about code - Refactoring is easier - Testing is easier - Separate pure code from impure code - Potential compiler optimizations (more in Haskell than in Scala)
(e.g. memoization, parallelisation, compute expressions at compile time) .footnote[ "What Referential Transparency can do for you"
Talk by Luka Jacobowitz at ScalaIO 2017
https://www.youtube.com/watch?v=X-cEGEJMx_4 ] --- ## *func*
is not
referentially transparent!
```scala def func(ioa1: Unit, ioa2: Unit): Unit = { ioa1 ioa2 } func(println("hi"), println("hi")) // prints "hi" twice //=> hi //=> hi println("-----") val x: Unit = println("hi") func(x, x) // prints "hi" once //=> hi ``` --- ## *func*
is
referentially transparent!
```scala def putStrLn(line: String): IO[Unit] = IO.eval { println(line) } def func(ioa1: IO[Unit], ioa2: IO[Unit]): IO[Unit] = for { _ <- ioa1 _ <- ioa2 } yield () func(putStrLn("hi"), putStrLn("hi")).run() // prints "hi" twice //=> hi //=> hi println("-----") val x: IO[Unit] = putStrLn("hi") func(x, x).run() // prints "hi" twice //=> hi //=> hi ``` --- class: middle, center name: IOMonad # 2. IO Monad
(My Impl)
--- ## Impure IO Program
with
side effects
```scala // impure program def program(): Unit = { print("Welcome to Scala! What's your name? ") val name = scala.io.StdIn.readLine println(s"Well hello, $name!") } ``` -- ```scala program() ```
-- - Whenever a method or a function returns *Unit* it is
***impure***
(or it is a noop). It's intension is to produce a side effect. - A
***pure***
function always returns a value of some type
(and doesn't produce a side effect inside). --- ## Pure IO Program
without
side effects
```scala // pure program val program: () => Unit = // () => Unit is syntactic sugar for: Function0[Unit] () => { print("Welcome to Scala! What's your name? ") val name = scala.io.StdIn.readLine println(s"Well hello, $name!") } ``` -- ```scala program() // producing the side effects "at the end of the world" ```
-- - Make the program a function returning *Unit*: *Function0[Unit]* - Free of side effects in it's definition - Produces side effects only when run (at the end of the world) - *program* can be a
val
(if no parameters are passed to it) --- ## Wrap Function0[A] in a case class
```scala case class IO[A](run: () => A) ``` -- ```scala // pure program val program: IO[Unit] = IO { () => { print("Welcome to Scala! What's your name? ") val name = scala.io.StdIn.readLine println(s"Well hello, $name!") } } ``` -- ```scala program.run() // producing the side effects "at the end of the world" ```
-- - *IO[A]* wraps a *Function0[A]* in a case class. - This is useful to implement further extensions on that case class. --- ## *IO#map* and *IO#flatMap*
```scala case class IO[A](run: () => A) { def flatMap[B](f: A => IO[B]): IO[B] = IO { () => f(run()).run() } def map[B](f: A => B): IO[B] = IO { () => f(run()) } def flatten[B](implicit ev: A <:< IO[B]): IO[B] = flatMap(a => a) } ``` -- ```scala val program: IO[Unit] = for { _ <- IO { () => print(s"Welcome to Scala! What's your name? ") } name <- IO { () => scala.io.StdIn.readLine } _ <- IO { () => println(s"Well hello, $name!") } } yield () ``` -- ```scala program.run() // producing the side effects "at the end of the world" ```
-- - With *map* and *flatMap* *IO[A]* is monadic (but it is not yet a Monad). - *IO* is ready for for-comprehensions. - This allows the composition of programs from smaller components. --- ## Companion object *IO* with *pure* and *eval* ```scala case class IO[A](run: () => A) { def flatMap[B](f: A => IO[B]): IO[B] = IO { () => f(run()).run() } def map[B](f: A => B): IO[B] = flatMap(a => pure(f(a))) def flatten[B](implicit ev: A <:< IO[B]): IO[B] = flatMap(a => a) } object IO { def pure[A](value: A): IO[A] = IO { () => value } // eager def eval[A](thunk: => A): IO[A] = IO { () => thunk } // lazy } ``` -- ```scala val program: IO[Unit] = for { welcome <- IO.pure("Welcome to Scala!") _ <- IO.eval { print(s"$welcome What's your name? ") } name <- IO.eval { scala.io.StdIn.readLine } // simpler with IO.eval _ <- IO.eval { println(s"Well hello, $name!") } } yield () ``` -- ```scala program.run() // producing the side effects "at the end of the world" ``` -- - *IO.pure* is
eager
and accepts a pure value. - *IO.eval* is
lazy
and accepts a computation. - *map* can be written in terms of *flatMap* and *pure*. --- ## Monad instance for *IO*
```scala sealed trait IO[+A] { def flatMap[B](f: A => IO[B]): IO[B] = IO { f(run()).run() } def map[B](f: A => B): IO[B] = flatMap(a => pure(f(a))) } object IO { implicit val ioMonad: Monad[IO] = new Monad[IO] { override def pure[A](value: A): IO[A] = IO.pure(value) override def flatMap[A, B](fa: IO[A])(f: A => IO[B]): IO[B] = fa flatMap f } } ``` --
- Monad instance defined in companion object (implicit scope) ---
### Computations that abstract over HKT: *F[_]: Monad*
```scala import scala.language.higherKinds import cats.syntax.flatMap._ import cats.syntax.functor._ def sumF[F[_]: Monad](from: Int, to: Int): F[Int] = Monad[F].pure { sumOfRange(from, to) } def fibonacciF[F[_]: Monad](num: Int): F[BigInt] = Monad[F].pure { fibonacci(num) } def factorialF[F[_]: Monad](num: Int): F[BigInt] = Monad[F].pure { factorial(num) } def computeF[F[_]: Monad](from: Int, to: Int): F[BigInt] = for { x <- sumF(from, to) y <- fibonacciF(x) z <- factorialF(y.intValue) } yield z ``` - This code can be used with *IO* or any other Monad. --- - **Reify *F[_] : Monad* with *IO*** ```scala val io: IO[BigInt] = computeF[IO](1, 4) val result: BigInt = io.run() println(result) //=><PHONE_NUMBER> ``` -- - **Reify *F[_] : Monad* with *cats.Id*** ```scala val result: cats.Id[BigInt] = computeF[cats.Id](1, 4) println(result) //=><PHONE_NUMBER> ``` -- - **Reify *F[_] : Monad* with *Option*** ```scala import cats.instances.option._ val maybeResult: Option[BigInt] = computeF[Option](1, 4) maybeResult foreach println //=><PHONE_NUMBER> ``` -- - **Reify *F[_] : Monad* with *Future*** ```scala import scala.concurrent.{Future, ExecutionContext} import ExecutionContext.Implicits.global import cats.instances.future._ val future: Future[BigInt] = computeF[Future](1, 4) future foreach println //=><PHONE_NUMBER> ``` --- ## Sync and async run* methods ```scala case class IO[A](run: () => A) { // ----- sync run* methods def runToTry: Try[A] = Try { run() } def runToEither: Either[Throwable, A] = runToTry.toEither // ----- async run* methods (all take an EC) def runToFuture(implicit ec: ExecutionContext): Future[A] = Future { run() } def runOnComplete(callback: Try[A] => Unit) (implicit ec: ExecutionContext): Unit = runToFuture onComplete callback def runAsync(callback: Either[Throwable, A] => Unit) (implicit ec: ExecutionContext): Unit = runOnComplete(tryy => callback(tryy.toEither)) } ``` --- class: middle, center, large-font The extended version of my IO Monad impl can be found here: https://github.com/hermannhueck/implementing-io-monad --- class: middle, center, large-font ## Monix Task is the
IO Monad
... a bit more sophisticated than my implementation. ;-) But it can also be seen as ... ## a
lazy Future
. --- class: middle, center name: FutureReferentiallyTransparent # 3. Is *Future* referentially transparent?
--- ## Is *Future* referentially transparent? ```scala val future1: Future[(Int, Int)] = { val atomicInt = new AtomicInteger(0) val future: Future[Int] = Future { atomicInt.incrementAndGet } for { x <- future y <- future } yield (x, y) } future1 onComplete println // Success((1,1)) ``` ```scala // same as future1, but inlined val future2: Future[(Int, Int)] = { val atomicInt = new AtomicInteger(0) for { x <- Future { atomicInt.incrementAndGet } y <- Future { atomicInt.incrementAndGet } } yield (x, y) } future2 onComplete println // Success((1,2)) <-- not the same result ``` -- ## No! --- ## Is my IO Monad referentially transparent? ```scala val io1: IO[(Int, Int)] = { val atomicInt = new AtomicInteger(0) val io: IO[Int] = IO.eval { atomicInt.incrementAndGet } for { x <- io y <- io } yield (x, y) } io1.runToFuture onComplete println // Success((1,2)) ``` ```scala // same as io1, but inlined val io2: IO[(Int, Int)] = { val atomicInt = new AtomicInteger(0) for { x <- IO.eval { atomicInt.incrementAndGet } y <- IO.eval { atomicInt.incrementAndGet } } yield (x, y) } io2.runToFuture onComplete println // Success((1,2)) <-- same result ``` -- ## Yes! --- ## Is Monix *Task* referentially transparent? ```scala val task1: Task[(Int, Int)] = { val atomicInt = new AtomicInteger(0) val task: Task[Int] = Task { atomicInt.incrementAndGet } for { x <- task y <- task } yield (x, y) } task1 runAsync println // Success((1,2)) ``` ```scala // same as task1, but inlined val task2: Task[(Int, Int)] = { val atomicInt = new AtomicInteger(0) for { x <- Task { atomicInt.incrementAndGet } y <- Task { atomicInt.incrementAndGet } } yield (x, y) } task2 runAsync println // Success((1,2)) <-- same result ``` -- ## Yes! --- class: middle, center name: WhatIsMonixTask # 4. What is Monix Task?
--- ## Monix 3.x
- library for asynchronous, reactive computations in Scala - available for the JVM and ScalaJS - dependent on: [*cats*](https://typelevel.org/cats/) and [*cats-effect*](https://typelevel.org/cats-effect/) - Web site: https://monix.io --- ## Monix 3.x abstractions:
- *monix.execution.
**Scheduler**
*: Scheduler that extends *scala.concurrrent.ExecutionContext* - *monix.eval.
**Task**
*: abstraction for async effects - *monix.eval.
**Coeval**
*: abstraction for sync effects - *monix.reactive.**Observable***: push-based streaming library
(compatible with the *reactivestreams.org* spec) - *monix.tail.
**Iterant**
*: pull-based streaming library
(compatible with the *reactivestreams.org* spec) --- ## Monix Task
- inspired by: Haskell's IO Monad, Scalaz Task - implementation of the IO Monad for Scala - lazy and referentially transparent alternative for *scala.concurrrent.Future* - *Future* is a value-wannabe. It starts executing when created. - *Task* is a function (a wrapped *Function0[A]*). It is just a description of a (possibly asynchronous) computation. It doesn't run unless explicitly told to do so. - good interop with *Future*: easy to turn *Future* to *Task* or *Task* to *Future* --- class: tiny-font ## Comparison of Features
Future
Task
Evaluation
eager / strict
lazy / non-strict
Control of side effects
no
yes
Memoization
always
possible, default: no
Referential Transparency
no
yes
Control of side effects
no
yes
Supports cancelation
no
yes
Supports blocking
yes
no (possible via Future)
Supports sync execution
no
yes
Runs on an other thread
always
not necessarily
Supports green threads
no
yes
Requires implicit ...
ExecutionContext
(for nearly every method)
Scheduler
(only when run)
Stack Safety
no
yes (due to trampolining)
Supports ScalaJS
yes
yes
--- class: middle, center name: TaskEvaluation # 5. Task Evaluation
--- ## Evaluation
Eager
Lazy
Synchronous
A
() => A
Eval[A], Coeval[A]
Asynchronous
(A => Unit) => Unit
() => (A => Unit) => Unit
Future[A]
Task[A], IO[A]
---
## build.sbt
```scala libraryDependencies += "io.monix" %% "monix" % "3.0.0-RC2" ``` --- ## Running a Task requires an implicit *Scheduler* -- Either ... ```scala import monix.execution.Scheduler.Implicits.global ``` or create an implicit instance ... ```scala import monix.execution.Scheduler.global implicit val scheduler: Scheduler = Scheduler.global ``` -- - *Scheduler* is a subclass of *ExecutionContext*, hence is serves as well as an *ExecutionContext* for *Future*. -- - The global *Scheduler* is backed by the global *ExecutionContext* which is backed by Java's *ForkJoinPool*. -- - *Future* requires an *ExecutionContext* for almost every method (*apply*, *map*, *flatMap*, *onComplete*, etc.) whereas ... -- - *Task* (due to it's laziness) requires a *Scheduler* only when run. --- ## Running *Future* and *Task* -- - *Future* is eager. It starts running when you create it. ```scala implicit val ec: ExecutionContext = ExecutionContext.global // needs EC to create Future val task: Future[Int] = Future { compute } // starts running HERE! val callback: Try[Int] => Unit = { // Try based callback case Success(result) => println(s"result = $result") case Failure(ex) => println(s"ERROR: ${ex.getMessage}") } task onComplete callback ``` -- - *Task* is a function, hence lazy. By it's creation nothing is run. ```scala val task: Task[Int] = Task { compute } val callback: Either[Throwable, Int] => Unit = { // Either based callback case Right(result) => println(s"result = $result") case Left(ex) => println(s"ERROR: ${ex.getMessage}") } implicit val scheduler: Scheduler = Scheduler.global // needs Scheduler to run Task task runAsync callback // starts running HERE! ``` --- ## Two flavours of *Callback* -- - Both are *Either* based. ```scala val callback: Either[Throwable, Int] => Unit = { case Right(result) => println(s"result = $result") case Left(ex) => println(s"ERROR: ${ex.getMessage}") } ``` ```scala val callback: Callback[Throwable, Int] = new Callback[Throwable, Int] { def onSuccess(result: Int): Unit = println(s"result = $result") def onError(ex: Throwable): Unit = println(s"ERROR: ${ex.getMessage}") } ``` -- - A *Callback* is a subclass of a function of type (Either[E, A] => Unit). ```scala package monix.execution abstract class Callback[-E, -A] extends (Either[E, A] => Unit) { def onSuccess(value: A): Unit def onError(e: E): Unit // ... } ``` --- ## Converting *Task* to *Future* - *Task#runToFuture* turns a *Task* into a *Future* and runs it. - A *Scheduler* is required in implicit scope. -- ```scala val task: Task[Int] = Task { compute } implicit val scheduler: Scheduler = Scheduler.global val future: Future[Int] = task.runToFuture val callback: Try[Int] => Unit = { // Try based callback case Success(result) => println(s"result = $result") case Failure(ex) => println(s"ERROR: ${ex.getMessage}") } future onComplete callback ``` --- ## *Task* cannot be blocked (directly) - The API of *Task* doesn't support any means to wait for a result. - But (if really needed) we can convert *Task* to *Future* and wait for the Future's result to become available. -- ```scala val task: Task[Int] = Task { compute }.delayExecution(1.second) implicit val scheduler: Scheduler = Scheduler.global val future: Future[Int] = task.runToFuture val result = Await.result(future, 3.seconds) println(s"result = $result") ``` --- ## *Task#foreach* - corresponds to *Future#foreach - *Task#foreach* runs the *Task*. - It processes the result, but ignores possible errors. -- ```scala val task: Task[Int] = Task { compute // returns an Int or throws an exception } implicit val scheduler: Scheduler = Scheduler.global task foreach { result => println(s"result = $result") } // prints computed value in case of success // prints nothing in case of error ``` - Use *foreach* only if the *Task* is guaranteed not to throw an *Exception*. --- ## *Task#failed* - corresponds to *Future#failed - Returns a failed projection of the current *Task*. - If the source fails we get a *Task[Throwable]* containing the error.
If the source *Task* succeeds we get a *NoSuchElementException*. -- ```scala val task: Task[Int] = Task { throw new IllegalStateException("illegal state") } val failed: Task[Throwable] = task.failed implicit val scheduler: Scheduler = Scheduler.global task foreach println // prints nothing failed foreach println // prints: java.lang.IllegalStateException: illegal state ``` --- class: middle, center name: TaskCancelation # 6. Task Cancelation
--- ## Canceling a *Task* - *Future* cannot be canceled, but *Task* can. -- ```scala val task: Task[Int] = Task { println("side effect") compute }.delayExecution(1.second) val callback: Either[Throwable, Int] => Unit = { case Right(result) => println(s"result = $result") case Left(ex) => println(s"ERROR: ${ex.getMessage}") } implicit val scheduler: Scheduler = Scheduler.global val cancelable: Cancelable = task runAsync callback // If we change our mind... cancelable.cancel() ``` --- ## *Cancelable* & *CancelableFuture* A *Cancelable* is a trait with a *cancel* method. ```scala package monix.execution trait Cancelable extends Serializable { def cancel(): Unit } ``` -- - *CancelableFuture* is a *(Future with Cancelable)*, hence a *Future* augmented with a *cancel* method. ```scala package monix.execution sealed abstract class CancelableFuture[+A] extends Future[A] with Cancelable { // ... } ``` --- ## Canceling a *Future* - *Task#runToFuture* returns a *CancelableFuture*. -- ```scala val task: Task[Int] = Task { compute }.delayExecution(1.second) implicit val scheduler: Scheduler = Scheduler.global val future: CancelableFuture[Int] = task.runToFuture future onComplete { case Success(result) => println(s"result = $result") case Failure(ex) => println(s"ERROR: ${ex.getMessage}") } // If we change our mind... future.cancel() ``` --- class: middle, center name: Memoization # 7. Memoization
--- ## Memoization - Overview -- ```scala // non-strict + memoized evaluation - equivalent to a lazy val val task = Task.eval { println("side effect"); "some value" } // same as apply val memoized = task.memoize ``` -- ```scala // non-strict + memoized evaluation - equivalent to a lazy val Task.evalOnce { println("side effect"); "memoized" } ``` -- ```scala // memoized only if successful val task = Task.eval { println("side effect"); "some value" } val memoizedIfSuccessful = task.memoizeOnSuccess ``` -- .footnote[ [Skip details](#TaskBuilders) ] --- ## *Task#memoize* - memoizes (caches) the *Task*'s result on the first run (like a *lazy val*) - is impure, as memoization is a side effect - guarantees idempotency -- ```scala val task = Task { println("side effect") fibonacci(20) } val memoized = task.memoize memoized runAsync printCallback //=> side effect //=> 10946 // Result was memoized on the first run! memoized runAsync printCallback //=> 10946 ``` --- ## *Task.evalOnce* - evaluates a thunk lazily on the first run and memoizes/caches the result - is impure, as memoization is a side effect - guarantees idempotency - *Task.evalOnce { thunk }* <-> *Task { thunk }.memoize* -- ```scala val task = Task.evalOnce { println("side effect") fibonacci(20) } implicit val scheduler: Scheduler = Scheduler.global task runAsync printCallback //=> side effect //=> 10946 // Result was memoized on the first run! task runAsync printCallback //=> 10946 ``` --- ## *Task#memoizeOnSuccess* - memoizes (caches) the *Task*'s result on the first run only if it succeeds - is impure, as memoization is a side effect -- ```scala var effect = 0 val source = Task.eval { println("side effect") effect += 1 if (effect < 3) throw new RuntimeException("dummy") else effect } val cached = source.memoizeOnSuccess cached runAsync printCallback //=> java.lang.RuntimeException: dummy cached runAsync printCallback //=> java.lang.RuntimeException: dummy cached runAsync printCallback //=> 3 cached runAsync printCallback //=> 3 ``` --- class: middle, center name: TaskBuilders # 8. Task Builders
--- ## Task Builders - Overview -- ```scala // non-strict evaluation - equivalent to a function Task.eval { println("side effect"); "always" } ``` -- ```scala // strict evaluation - lifts a pure value into Task context Task.now { println("side effect"); "immediate" } ``` -- ```scala // strict evaluation - lifts a Throwable into Task context Task.raiseError { println("side effect"); new Exception } ``` -- ```scala // non-strict + memoized evaluation - equivalent to a lazy val Task.evalOnce { println("side effect"); "memoized" } ``` -- ```scala // non-strict evaluation - turns a (possibly eager) Task into a lazy one Task.defer { println("side effect"); Task.now("no longer immediate") } ``` -- ```scala // non-strict evaluation on a different "logical" thread Task.evalAsync { println("side effect"); "always on different thread" } ``` -- .footnote[ [Skip details](#FutureToTask) ] --- name: TaskEval ## *Task.eval* / *Task.apply* / *Task.delay* - corresponds to *Future.apply* - evaluates a thunk lazily -- ```scala val task = Task.eval { println("side effect") fibonacci(20) } implicit val scheduler: Scheduler = Scheduler.global task foreach println //=> side effect //=> 10946 // The evaluation (and thus all contained side effects) // gets triggered on each run: task runAsync printCallback //=> side effect //=> 10946 ``` --- name: TaskNow ## *Task.now* / *Task.pure* - corresponds to *Future.successful* - lifts a pure value into the *Task* context - evaluates eagerly / immediately, never spawning a separate thread - !!! Don't use it for computations and side effects, only for pure values !!! -- ```scala val task = Task.now { println("(eagerly produced) side effect; DON'T DO THAT !!") 42 } //=> (eagerly produced) side effect; DON'T DO THAT !! implicit val scheduler: Scheduler = Scheduler.global task runAsync printCallback //=> 42 ``` --- name: TaskUnit ## *Task.unit* - corresponds to *Future.unit* - lifts a pure Unit value into the *Task* context - evaluates eagerly / immediately, never spawning a separate thread - Task.unit <-> Task.now(()) -- ```scala val task: Task[Unit] = Task.unit implicit val scheduler: Scheduler = Scheduler.global task runAsync println //=> Right(()) ``` --- name: TaskRaiseError ## *Task.raiseError* - corresponds to *Future.failed* - lifts a *Throwable* into the *Task* context (analogous to *Task.now*) - evaluates eagerly / immediately, never spawning a separate thread - !!! Don't use it for computations and side effects, only for pure errors/exceptions !!! -- ```scala val task = Task.raiseError { println("(eagerly produced) side effect; DON'T DO THAT !!") new IllegalStateException("illegal state") } //=> (eagerly produced) side effect; DON'T DO THAT !! implicit val scheduler: Scheduler = Scheduler.global task runAsync printCallback //=> java.lang.IllegalStateException: illegal state ``` --- name: TaskEvalOnce ## *Task.evalOnce* - corresponds to *Future.apply* - evaluates a thunk lazily on the first run and memoizes/caches the result - is impure, as memoization is a side effect - guarantees idempotency - Task.evalOnce <--> Task.eval.memoize -- ```scala val task = Task.evalOnce { println("side effect") fibonacci(20) } implicit val scheduler: Scheduler = Scheduler.global task foreach println //=> side effect //=> 10946 // Result was memoized on the first run! task runAsync printCallback //=> 10946 ``` --- name: TaskNever ## *Task.never* - corresponds to *Future.never* - creates a *Task* that never completes - evaluates lazily -- ```scala val never: Task[Int] = Task.never[Int] val timedOut: Task[Int] = never.timeoutTo(3.seconds, Task.raiseError(new TimeoutException)) implicit val scheduler: Scheduler = Scheduler.global timedOut runAsync println // After 3 seconds: // => Left(java.util.concurrent.TimeoutException) ``` --- name: TaskDefer ## *Task.defer* / *Task.suspend* - takes a (possibly eager) *Task* and turns it into a lazy *Task* of the same type. - evaluates lazily - *Task.defer(Task.now(value))* <-> Task.eval(value) -- ```scala val task = Task.defer { println("side effect") Task.now(42) } implicit val scheduler: Scheduler = Scheduler.global task runAsync printCallback //=> side effect //=> 10946 // The evaluation (and thus all contained side effects) // gets triggered on each run: task runAsync printCallback //=> side effect //=> 10946 ``` --- name: TaskEval ## *Task.evalAsync* - evaluates a thunk lazily like *Task.eval* - guaranteed to spawn a separate "logical" thread - Task.evalAsync(a) <-> Task.eval(a).executeAsync - See also: [Async Boundaries](#AsyncBoundaries) -- ```scala val task = Task.evalAsync { println("side effect") fibonacci(20) } implicit val scheduler: Scheduler = Scheduler.global // The evaluation (and thus all contained side effects) // gets triggered on each run. // But it is run asynchronously on a different "logical" thread. task runAsync printCallback //=> side effect //=> 10946 ``` --- class: middle, center name: FutureToTask # 9. Converting Future to Task
--- ## *Future* to *Task* - Overview -- ```scala // --- SOME LIBRARY --- private def compute: Int = ??? // compute Int value // compute Int value async - library code returns a Future def computeAsync(implicit ec: ExecutionContext): Future[Int] = Future { compute } ``` -- ```scala // converts a Future to a effectively memoized Task. Strict evaluation !!! import monix.execution.Scheduler.Implicits.global val task = Task.fromFuture(computeAsync) // needs an EC ``` -- ```scala // converts a Future to a lazy Task.  Non-strict evaluation !!! import monix.execution.Scheduler.Implicits.global val task = Task.deferFuture(computeAsync) // needs an EC ``` -- ```scala // converts a Future to a lazy Task. Non-strict evaluation !!! val task = Task.deferFutureAction(implicit scheduler => computeAsync) // Scheduler is required when the task is run. ``` -- [Skip details](#TailRecursiveLoops) --- ## *Task.fromFuture* - converts a *Future* to an effectively memoized *Task*. **Strict** evaluation !!! - The *Future* starts running eagerly, before *Task.fromFuture* is invoked. - An implicit *ExecutionContext* or *Scheduler* must be provided. -- ```scala object library { def futureFactorial(n: Int)(implicit ec: ExecutionContext): Future[BigInt] = Future { println("side effect"); factorial(n) } } import library._ import Scheduler.Implicits.global // Future created before Task.fromFuture is invoked // Hence the Future is already running before we convert it to a Task val task = Task.fromFuture(futureFactorial(10)) //=> side effect task foreach println //=> 3628800 ``` --- ## *Task.defer* & *Task.fromFuture* - converts a *Future* into a lazy *Task*. **Non-strict**/lazy evaluation !!! - *Task.defer* effectively creates a function which - when invoked - will create the *Future*. - The *Future* doesn't start running (unless created outside *Task.defer*). - An implicit *ExecutionContext* or *Scheduler* must be provided. -- ```scala object library { def futureFactorial(n: Int)(implicit ec: ExecutionContext): Future[BigInt] = Future { println("side effect"); factorial(n) } } import library._ import Scheduler.Implicits.global val task = Task.defer { Task.fromFuture(futureFactorial(10)) } task foreach println //=> side effect //=> 3628800 ``` --- ## *Task.deferFuture* - converts a *Future* into a lazy *Task*. **Non-strict**/lazy evaluation !!! - *Task.deferFuture(f)* <-> *Task.defer(Task.fromFuture(f))* - The *Future* doesn't start running (unless created outside *Task.deferFuture*).. - An implicit *ExecutionContext* or *Scheduler* must be provided. -- ```scala object library { def futureFactorial(n: Int)(implicit ec: ExecutionContext): Future[BigInt] = Future { println("side effect"); factorial(n) } } import library._ import Scheduler.Implicits.global val task = Task.deferFuture(futureFactorial(10) task foreach println //=> side effect //=> 3628800 ``` --- ## *Task.deferFutureAction* - converts a *Future* into a lazy *Task*. **Non-strict**/lazy evaluation !!! - The *Future* doesn't start running (unless created outside *Task.deferFutureAction*).. - **No** implicit *ExecutionContext* or *Scheduler* must be provided. - An implicit *Scheduler* must be provided when the task is run. -- ```scala object library { def futureFactorial(n: Int)(implicit ec: ExecutionContext): Future[BigInt] = Future { println("side effect"); factorial(n) } } import library._ val task = Task.deferFutureAction(implicit scheduler => futureFactorial(10)) import Scheduler.Implicits.global task foreach println //=> side effect //=> 3628800 ``` --- class: middle, center name: TaskAsMonad # 10. Task as Monad
--- ## *Task#map*
```scala val task: Task[List[(String, Int)]] = Task.eval { // Attention! This Task doesn't close the resource "README.md" Source.fromURL("file:./README.md") } map { _.getLines } map { _.toList } map { wordCount(limit = 2) } ``` ```scala import Scheduler.Implicits.global task runAsync new Callback[Throwable, List[(String, Int)]] { def onError(ex: Throwable): Unit = println(s"ERROR: ${ex.toString}") def onSuccess(wcList: List[(String, Int)]): Unit = wcList foreach { case (word, count) => println(s"$word: $count")} } ``` --- ## *Task#flatMap*
```scala def sumTask(from: Int, to: Int) = Task { sumOfRange(from, to) } def fibonacciTask(num: Int) = Task { fibonacci(num) } def factorialTask(num: Int) = Task { factorial(num) } def computeTask(from: Int, to: Int): Task[BigInt] = sumTask(from, to) .flatMap(fibonacciTask) .map(_.intValue) .flatMap(factorialTask) ``` ```scala val task = computeTask(1, 4) import Scheduler.Implicits.global task runAsync printCallback //=><PHONE_NUMBER> ``` --- ## For-comprehension over a *Task* context
```scala def sumTask(from: Int, to: Int) = Task { sumOfRange(from, to) } def fibonacciTask(num: Int) = Task { fibonacci(num) } def factorialTask(num: Int) = Task { factorial(num) } def computeTask(from: Int, to: Int): Task[BigInt] = for { x <- sumTask(from, to) y <- fibonacciTask(x) z <- factorialTask(y.intValue) } yield z ``` ```scala val task = computeTask(1, 4) import Scheduler.Implicits.global task runAsync printCallback //=><PHONE_NUMBER> ``` --- ## *Task* has a *Monad* instance ```scala def sumF[F[_]: Monad](from: Int, to: Int): F[Int] = Monad[F].pure { sumOfRange(from, to) } def fibonacciF[F[_]: Monad](num: Int): F[BigInt] = Monad[F].pure { fibonacci(num) } def factorialF[F[_]: Monad](num: Int): F[BigInt] = Monad[F].pure { factorial(num) } def computeF[F[_]: Monad](from: Int, to: Int): F[BigInt] = for { x <- sumF(from, to) y <- fibonacciF(x) z <- factorialF(y.intValue) } yield z ``` ```scala // reify F[] with Task val task: Task[BigInt] = computeF[Task](1, 4) import Scheduler.Implicits.global task runAsync printCallback //=><PHONE_NUMBER> ``` --- ## *computeF* can be used with any *Monad* instance ```scala import scala.concurrent.Future import scala.concurrent.ExecutionContext.Implicits.global import cats.instances.future._ // reify F[] with Future val task: Future[BigInt] = computeF[Future](1, 4) task foreach { result => println(s"result = $result")} //=><PHONE_NUMBER> ``` ```scala import cats.Id // reify F[] with Id val result: Id[BigInt] = computeF[Id](1, 4) // Id[BigInt] ^= BigInt println(s"result = $result") //=><PHONE_NUMBER> ``` ```scala import cats.instances.option._ // reify F[] with Option val maybeResult: Option[BigInt] = computeF[Option](1, 4) maybeResult foreach { result => println(s"result = ${result}") } //=><PHONE_NUMBER> ``` --- class: middle, center name: TailRecursiveLoops # 11. Tail Recursive Loops
--- ## Tail recursion with annotation *@tailrec* - With the *@tailrec* annaotation the compiler guarantees us stack-safety. ```scala @scala.annotation.tailrec def fibonacci(cycles: Int, x: BigInt = 0, y: BigInt = 1): BigInt = if (cycles > 0) fibonacci(cycles-1, y, x + y) else y ``` ```scala val result: BigInt = fibonacci(6) println(result) //=> 13 ``` --- ## Tail recursion with *Task* - With *Task* the fibonacci impl is lazy (due to *Task.defer*). - The annotation *@tailrec* is no longer required. - The impl is nevertheless stack-safe. ```scala def fibonacciTask(cycles: Int, x: BigInt = 0, y: BigInt = 1): Task[BigInt] = if (cycles > 0) Task.defer(fibonacciTask(cycles-1, y, x+y)) else Task.now(y) ``` ```scala val task: Task[BigInt] = fibonacciTask(6) implicit val scheduler: Scheduler = Scheduler.global task foreach println // fibonacci computation starts here //=> 13 ``` --- ## Tail recursion with *Task#flatMap* - With *Task* the fibonacci impl is lazy (due to *Task#flatMap*). - The annotation *@tailrec* is no longer required. - The impl is nevertheless stack-safe. ```scala def fibonacciTask(cycles: Int, x: BigInt = 0, y: BigInt = 1): Task[BigInt] = Task.eval(cycles > 0).flatMap { case true => fib(cycles-1, y, x+y) case false => Task.now(y) } ``` ```scala val task: Task[BigInt] = fibonacciTask(6) implicit val scheduler: Scheduler = Scheduler.global task foreach println // fibonacci computation starts here //=> 13 ``` --- ## Mutual tail recursion - With *Task* even mutual tail recursive calls are possible. ```scala def odd(n: Int): Task[Boolean] = Task.eval(n == 0).flatMap { case true => Task.now(false) case false => if (n > 0) even(n - 1) else even(n + 1) } def even(n: Int): Task[Boolean] = Task.eval(n == 0).flatMap { case true => Task.now(true) case false => if (n > 0) odd(n - 1) else odd(n + 1) } ``` ```scala val task: Task[Boolean] = even(-1000000) implicit val scheduler: Scheduler = Scheduler.global task foreach println //=> true ``` --- class: middle, center name: AsyncBoundaries # 12. Async Boundaries
--- ## Green Threads and JVM Threads 1.
JVM threads
map 1:1 to OS threads. They have their own stack and register set each. They are managed by the OS using
preemptive multitasking
in kernel space. JVM threads are a scarce resource. Switching context between them is expensive. 2.
[Green threads](https://en.wikipedia.org/wiki/Green_threads)
(aka *Fibers*) are managed by the runtime (Haskell) or a library (Monix, Cats Effect, Akka). This allows for
[cooperative multitasking](https://en.wikipedia.org/wiki/Cooperative_multitasking)
in user space. Green threads can yield execution back to the thread pool which then might choose another green thread to run on the same JVM thread. A context switch between green threads is cheap. Your application may use relatively few JVM threads but many green threads. Monix *Task* provides asynchrony with JVM threads and green threads. - Avoid blocking a JVM thread! - You may block a green thread without blocking the JVM thread.
("asynchronous blocking" or "semantic blocking"). --- ## Asynchronous Boundaries - Overview - An asynchronous boundary may be a context switch between two logical threads. - It may also mean that the current thread is rescheduled on the current thread pool. -- ```scala // Task#executeAsync ensures an async boundary, forks into another logical thread. val source = Task.eval { computation } val forked = source.executeAsync ``` -- ```scala // Task#executeOn executes the current Task on another Scheduler. // This implies a context switch between JVM threads. val source = Task.eval { computation } val forked = source.executeOn(someScheduler) ``` -- ```scala // Task#asyncBoundary introduces an async Boundary // (possibly switching back to the default Scheduler). val source = Task.eval { computation } val forked = source.executeOn(someScheduler) val back = forked.asyncBoundary ``` -- [Skip details](#Schedulers) --- ## *Task#executeAsync* - ensures an asynchronous boundary - forces the fork into another thread on execution -- ```scala val task = Task.eval { println(s"Running on $currentThread") factorial(10) }.executeAsync import Scheduler.Implicits.global task.runToFuture foreach println //=> Running on Thread: scala-execution-context-global-241 //=> 3628800 ``` --- ## *Task#executeOn* (1/2) - executes the current Task on another Scheduler - (possibly) introduces an asynchronous boundary - (possibly) implies a context switch between JVM threads -- ```scala lazy val io = Scheduler.io(name="my-io") // I/O scheduler val source = Task { println(s"Running on $currentThread") } val forked = source.executeOn(io) // override default Scheduler by fork import Scheduler.Implicits.global source.runToFuture //=> Running on Thread: run-main-9 forked.runToFuture //=> Running on Thread: my-io-265 ``` --- ## *Task#executeOn* (2/2) ```scala lazy val io = Scheduler.io(name="my-io") // I/O scheduler val source = Task { println(s"Running on $currentThread") } val forked = source.executeOn(io) // override default Scheduler by fork val onFinish = Task { println(s"Ending on $currentThread") } val task = source .flatMap(_ => forked) .doOnFinish(_ => onFinish) import Scheduler.Implicits.global task.runToFuture //=> Running on thread: run-main-2a //=> Running on thread: my-io-803 //=> Ending on thread: scala-execution-context-global-800 ``` --- ## *Task#asyncBoundary* - switches execution back to the default scheduler - introduces an asynchronous boundary (if task is already running on the default scheduler) -- ```scala lazy val io = Scheduler.io(name="my-io") // I/O scheduler val source = Task { println(s"Running on $currentThread") } val forked = source.executeOn(io) // override default Scheduler by fork val onFinish = Task { println(s"Ending on $currentThread") } val task = source // executes on global .flatMap(_ => forked) // executes on io .asyncBoundary // switch back to global .doOnFinish(_ => onFinish) // executes on global import Scheduler.Implicits.global task.runToFuture //=> Running on Thread: run-main-e //=> Running on Thread: my-io-358 //=> Ending on Thread: scala-execution-context-global-343 ``` --- ## Asynchrony on a single JVM Thread - Green threads allow for asynchrony on a single JVM thread. - The example *Task* introduces several async boundaries which allow to shift the context of logical threads. - The Task is run twice on a thread pool with a single thread. -- ```scala def createTask(id: Int): Task[Int] = Task { println(s"Task no $id starting on $currentThread"); id } .asyncBoundary .map { someComputation } .asyncBoundary .map { someComputation } .asyncBoundary .map { someComputation } .asyncBoundary .map { someComputation } implicit val scheduler: Scheduler = Scheduler.singleThread("SingleThreadScheduler") createTask(1) runAsync (_ => println(s"Task no 1 finished")) createTask(2) runAsync (_ => println(s"Task no 2 finished")) ``` ---
### Output ```code Task no 1 starting on Thread: run-main-1b Computation no 1 running on Thread: SingleThreadScheduler-587 Task no 2 starting on Thread: run-main-1b Computation no 2 running on Thread: SingleThreadScheduler-587 Computation no 1 running on Thread: SingleThreadScheduler-587 Computation no 2 running on Thread: SingleThreadScheduler-587 Computation no 1 running on Thread: SingleThreadScheduler-587 Computation no 2 running on Thread: SingleThreadScheduler-587 Computation no 1 running on Thread: SingleThreadScheduler-587 Task no 1 finished on Thread: SingleThreadScheduler-587 Computation no 2 running on Thread: SingleThreadScheduler-587 Task no 2 finished on Thread: SingleThreadScheduler-587 ``` --- class: middle, center name: Schedulers # 13. Schedulers
--- ## *ExecutionContext* - Downsides
*ExecutionContext* is limited because ... - It cannot execute things with a given delay. - It cannot execute units of work periodically (e.g. once every second). - The execute() method doesn’t return a token you could use to cancel the pending execution of a task.
*ExecutionContext* is **shared mutable state** ... and *Scheduler* is too. But there is a big difference: - A *Scheduler* is only needed when running a *Task* (at the end of the world). Whereas an *ExecutionContext* is already required when you create a *Future* and again for nearly every method of *Future* like *map*, *flatMap* and *onComplete*. --- ## Scheduler
- *Scheduler* is a subclass of *ExecutionContext*, hence it also serves as enhanced *ExecutionContext* for *Future*. - It is also a replacement for Java’s *ScheduledExecutorService* (due to it's scheduling and cancelation capabilities). - *Future* requires an *ExecutionContext* for almost every method (*apply*, *map*, *flatMap*, *onComplete*, etc.) whereas ... - *Task* (due to it's laziness) requires a *Scheduler* only when run. --- ## Run a *Runnable* on a *Scheduler*
- A *Scheduler* can execute *Runnable*s directly. ```scala Scheduler.global.execute(new Runnable { def run(): Unit = println("Hello, world!") }) ``` - We normally do not work with *Runnable*s, we use other abstractions like *Future* or *Task*. - Running a *Task* requires an implicit *Scheduler* in scope. Compared to *ExecutionContext* a *Scheduler* ... - can execute a *Runnable* with a delay (EC cannot). - can execute a *Runnable* periodically (EC cannot). - can provide a cancel token to cancel a running execution (EC cannot). --- ## Execute *Runnable*s - Overview
- *Scheduler* provides different ways to run *Runnable*s. -- ```scala // executes a Runnable scheduler.execute(runnable) ``` -- ```scala // schedules a thunk for execution with a delay val cancelable: Cancelable = scheduler.scheduleOnce(initialDelay) { thunk } ``` -- ```scala // schedules a thunk with an initial delay and a periodic delay // between the end of a task and the start of the next task. // The effective delay is periodic delay + duration of the task val cancelable: Cancelable = scheduler.scheduleWithFixedDelay( initialDelay, periodicDelay) { thunk } ``` -- ```scala // schedules a thunk with an initial delay and a delay between periodic restarts // The effective delay is only the periodic delay. // (The duration of the task is not taken into account.) val cancelable: Cancelable = scheduler.scheduleAtFixedRate( initialDelay, periodicDelay) { thunk } ``` --- ## Execution Models
- A Scheduler has an *ExecutionModel*. - The *ExecutionModel* affects **how** a *Task* is evaluated. - Monix 3.x provides three different *ExecutionModel*s:
SynchronousExecution
: preferres sync execution as long as possible
AlwaysAsyncExecution
: ensures async execution for each step of a task
BatchedExecution
(
default
): specifies a mixed execution mode under which tasks are executed synchronously in batches up to a maximum size, after which an asynchronous boundary is forced.
(The batch size is configurable. Default: 1024)
--- ## *UncaughtExceptionReporter* - A *Scheduler* has an *UncaughtExceptionReporter*. - *UncaughtExceptionReporter* is a trait with a method *reportFailure*: ```scala package monix.execution trait UncaughtExceptionReporter extends Serializable { def reportFailure(ex: Throwable): Unit } ``` - The default reporter logs exceptions to STDERR (on the JVM, not on ScalaJS). - Create your own customized *UncaughtExceptionReporter* if needed: ```scala val myReporter = UncaughtExceptionReporter(ex => log.error(ex)) implicit lazy val myScheduler = Scheduler(ExecutionContext.global, myReporter) ``` --- ## Build your own *Scheduler*
- The global *Scheduler* is backed by the global *ExecutionContext* which is backed by Java's *ForkJoinPool*. - The global *Scheduler* is a good default suitable for many scenarios. But ... - You can create your own *Scheduler* for the needs of your application. - Monix provides many builders to build your own.  - *SchedulerService* can be used to initiate and control the shutdown process.
shutdown
: Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted.
awaitTermination
: Returns a Future that will be complete when all tasks have completed execution after a shutdown request, or the timeout occurs.
isShutdown
: Returns true if this scheduler has been shut down.
isTerminated
: Returns true if all tasks have completed after shut down.
--- ## Shutdown with SchedulerService - Code ```scala val io: SchedulerService = Scheduler.io("my-io") io.execute(() => { Thread sleep 5000L println(s"Running on $currentThread") println("Hello, world!") }) io.shutdown() println(s"isShutdown = ${io.isShutdown}") val termination: Future[Boolean] = io.awaitTermination(10.seconds, global) Await.ready(termination, Duration.Inf) println(s"isTerminated = ${io.isTerminated}") ``` ### Output ```code isShutdown = true Running on Thread: my-io-114 Hello, world! isTerminated = true ``` --- class: middle, center name: ErrorHandling # 14. Error Handling
--- ## Signaling of Errors - Overview
- Errors in *Task*s are signaled in the result of a *Task*. You can react on errors in the callback (in *case Left(ex)* or in *case Failure(ex)*) - Errors in callbacks are logged to the *UncaughtExceptionReporter* of the *Scheduler* (which prints to stderr by default). - The *UncaughtExceptionReporter* of the *Scheduler* can be overriden in order to log to a different destination. --- ## Signaling of *Task* Errors
- Errors in the *Task* impl are signaled to the run* methods, e.g. in *runAsync* ```scala val task: Task[Int] = Task(Random.nextInt).flatMap { case even if even % 2 == 0 => Task.now(even) case odd => throw new IllegalStateException(odd.toString) // Error in task definition } implicit val scheduler: Scheduler = Scheduler.global task runAsync println //=> Right(-924040280) task runAsync println //=> Left(java.lang.IllegalStateException: 834919637) ``` --- ## Signaling of Callback Errors
- Errors in the callback are logged to the *UncaughtExceptionReporter* (stderr by default). ```scala // Ensures asynchronous execution, just to show that // the action doesn't happen on the current thread. val source = Task(sumOfRange(1, 100)) .delayExecution(10.seconds) implicit val scheduler: Scheduler = Scheduler.global task runAsync { r => throw new IllegalStateException(r.toString) } // After 1 second, this will log the whole stack trace: java.lang.IllegalStateException: Right(2) at chap14errorhandling.App02SignalingErrorInCallback$.$anonfun$new$1(App02SignalingErrorInCallback.scala:19) at monix.execution.Callback$$anon$3.onSuccess(Callback.scala:145) // stack trace continued ... ``` --- ## Overriding the Error Logging
- Override *UncaughtExceptionReporter* to log to a different log destination. ```scala import monix.execution.UncaughtExceptionReporter import org.slf4j.LoggerFactory val reporter = UncaughtExceptionReporter { ex => val logger = LoggerFactory.getLogger(this.getClass) // log with SLF4J logger.error("Uncaught exception", ex) } implicit val global: Scheduler = Scheduler(Scheduler.global, reporter) val source = Task(sumOfRange(1, 100)) .delayExecution(10.seconds) task.runAsync { r => throw new IllegalStateException(r.toString) } // logs Exception with SLF4J ``` --- ## Timouts - Overview
```scala // Triggers TimeoutException if the source does not complete in 3 seconds val timedOut: Task[Int] = source.timeout(3.seconds) ``` ```scala // Triggers a Fallback Task if the source does not complete in 3 seconds source.timeoutTo( 3.seconds, Task.raiseError(new TimeoutException("That took too long!")) // Fallback Task ) ``` --- ## *Task#timout*
- triggers a *TimoutException* after the specified time. ```scala val source: Task[Int] = Task(sumOfRange(1, 100)) .delayExecution(10.seconds) // Triggers TimeoutException if the source does not // complete in 3 seconds after runAsync val timedOut: Task[Int] = source.timeout(3.seconds) implicit val scheduler: Scheduler = Scheduler.global timedOut runAsync println //=> Left(java.util.concurrent.TimeoutException: Task timed-out after 3 seconds of inactivity) ``` --- ## *Task#timoutTo*
- falls back to another *Task* after the specified time. ```scala val source: Task[Int] = Task(sumOfRange(1, 100)) .delayExecution(10.seconds) // Triggers TimeoutException if the source does not // complete in 3 seconds after runAsync val timedOut: Task[Int] = source.timeoutTo( 3.seconds, Task.raiseError(new TimeoutException("That took too long!")) // Fallback Task ) implicit val scheduler: Scheduler = Scheduler.global timedOut runAsync println //=> Left(java.util.concurrent.TimeoutException: That took too long!) ``` --- ## Error Handlng - Overview (1/2)
```scala // handles failure with the specified (lazy) handler Function1 val f: Function1[Throwable, Task[B]] = ??? // error handler task.onErrorHandleWith(f) ``` ```scala // handles failure with the specified (lazy) handler PartialFunction val pf: PartialFunction[Throwable, Task[B]] = ??? // error handler task.onErrorRecoverWith(pf) ``` ```scala // handles failure with the specified (eager) handler Function1 val f: Function1[Throwable, B] = ??? // error handler task.onErrorHandle(f) ``` ```scala // handles failure with the specified (eager) handler PartialFunction val pf: PartialFunction[Throwable, B] = ??? // error handler task.onErrorRecoverWith(pf) ``` --- ## Error Handlng - Overview (2/2)
```scala // in case of failure keeps retrying until maxRetries is reached val f: Throwable => Task[B] = ??? // error handler task.onErrorRestart(maxRetries = 3) ``` ```scala // in case of failure keeps retrying until predicate p becomes true val p: Throwable => Boolean = ??? task.onErrorRestart(f) ``` --- ## *Task#onErrorHandleWith*
- provides a function *Throwable => Task[B]* which returns a (lazy) fallback *Task* in case an exception is thrown. ```scala val source = Task(sumOfRange(1, 100)) .delayExecution(10.seconds) .timeout(3.seconds) val recovered = source.onErrorHandleWith { case _: TimeoutException => // Oh, we know about timeouts, recover it Task.now("Recovered!") case other => // We have no idea what happened, raise error! Task.raiseError(other) } implicit val scheduler: Scheduler = Scheduler.global recovered.runToFuture foreach println //=> Recovered! ``` --- ## *Task#onErrorRecoverWith*
- provides a *PartialFunction[Throwable, Task[B]]* which returns a (lazy) fallback *Task* in case an exception is thrown. - avoids the 'other' case in the previous example. ```scala val source = Task(sumOfRange(1, 100)) .delayExecution(10.seconds) .timeout(3.seconds) val recovered = source.onErrorRecoverWith { case _: TimeoutException => // Oh, we know about timeouts, recover it Task.now("Recovered!") } implicit val scheduler: Scheduler = Scheduler.global recovered.runToFuture foreach println //=> Recovered! ``` --- ## *Task#onErrorHandle*
- provides a function *Throwable => B* which returns a (eager) fallback value in case an exception is thrown. ```scala val source = Task(sumOfRange(1, 100)) .delayExecution(10.seconds) .timeout(3.seconds) val recovered = source.onErrorHandle { case _: TimeoutException => // Oh, we know about timeouts, recover it "Recovered!" case other => throw other // Rethrowing } implicit val scheduler: Scheduler = Scheduler.global recovered.runToFuture foreach println //=> Recovered! ``` --- ## *Task#onErrorRecover*
- provides a *PartialFunction[Throwable, B]* which returns a (eager) fallback value in case an exception is thrown. - avoids the 'other' case in the previous example. ```scala val source = Task(sumOfRange(1, 100)) .delayExecution(10.seconds) .timeout(3.seconds) val recovered = source.onErrorRecover { case _: TimeoutException => // Oh, we know about timeouts, recover it "Recovered!" } implicit val scheduler: Scheduler = Scheduler.global recovered.runToFuture foreach println //=> Recovered! ``` --- ## *Task#onErrorRestart*
- specifies max. retries in case *Task* execution throws an exception. ```scala val source = Task(Random.nextInt).flatMap { case even if even % 2 == 0 => Task.now(even) case other => Task.raiseError(new IllegalStateException(other.toString)) } // Will retry 2 times for a random even number, or fail if the maxRetries is reached! val randomEven: Task[Int] = source.onErrorRestart(maxRetries = 3) implicit val scheduler: Scheduler = Scheduler.global randomEven runAsync println ``` --- ## *Task#onErrorRestartIf*
- keeps retrying in case of an exception as long as the specified predicate *Throwable => Boolean* is true. ```scala val source = Task(Random.nextInt).flatMap { case even if even % 2 == 0 => Task.now(even) case other => Task.raiseError(new IllegalStateException(other.toString)) } // Will keep retrying for as long as the source fails with an IllegalStateException val randomEven: Task[Int] = source.onErrorRestartIf { case _: IllegalStateException => true case _ => false } implicit val scheduler: Scheduler = Scheduler.global randomEven runAsync println ``` --- ## Retry with Backoff
- *onErrorHandleWith* allows us to implement a retry with backoff. ```scala val source = Task(Random.nextInt).flatMap { case even if even % 2 == 0 => Task.now(even) case other => Task.raiseError(new IllegalStateException(other.toString)) } def retryBackoff[A](source: Task[A], maxRetries: Int, firstDelay: FiniteDuration): Task[A] = source.onErrorHandleWith { case ex: Exception => if (maxRetries > 0) { println(s"Retrying ... maxRetries = $maxRetries, nextDelay = ${firstDelay * 2}") // Recursive call, it's OK as Monix is stack-safe retryBackoff(source, maxRetries - 1, firstDelay * 2).delayExecution(firstDelay) } else Task.raiseError(ex) } val randomEven: Task[Int] = retryBackoff(source, 3, 1.second) implicit val scheduler: Scheduler = Scheduler.global println(Await.result(randomEven.runToFuture, 10.seconds)) ``` --- ## Task#restartUntil
- restarts a *Task* until the specified predicate is true. ```scala val random = Task.eval(Random.nextInt()) val predicate: Int => Boolean = _ % 2 == 0 val randomEven = random.restartUntil(predicate) implicit val scheduler: Scheduler = Scheduler.global randomEven.runToFuture foreach println // prints an even Int number ``` --- ## Task#doOnFinish
- specifies an extra finish callback *Option[Throwable] => Task[Unit]* to be invoked when the *Task* is finished. ```scala val task: Task[Int] = Task(1) val finishCallback: Option[Throwable] => Task[Unit] = { case None => println("Was success!") Task.unit case Some(ex) => println(s"Had failure: $ex") Task.unit } val withFinishCallback: Task[Int] = task doOnFinish finishCallback implicit val scheduler: Scheduler = Scheduler.global withFinishCallback.runToFuture foreach println //=> Was success! //=> 1 ``` --- class: middle, center name: Races # 15. Races
--- ## Races - Overview
```scala // runs 2 Tasks in parallel. // the resulting Either allows // to cancel the other Task or to await its completion. val raceTask: Task[Either[(Int, Fiber[Int]), (Fiber[Int], Int)]] = Task.racePair(task1, task2) ``` ```scala // runs 2 Tasks in parallel and automatically cancels the losere Task. // the resulting Either gives access to the winner's result. val winnerTask: Task[Either[Int, Int]] = Task.race(task1, task2) ``` ```scala // runs a Seq[Task[A]] in parallel and automatically cancels the losers. // Running it returns the winners result. val winnerTask: Task[Int] = Task.raceMany(Seq(task1, task2)) ``` --- ## *Fiber*
- With a *Fiber* you can cancel the other (not yet completed) *Task* or join to it (= wait for it).
```scala trait Fiber[A] extends cats.effect.Fiber[Task, A] { // triggers the cancelation of the fiber. def cancel: CancelToken[Task] // returns a new task that will await the completion of the underlying fiber. def join: Task[A] } ``` --- ## *Task.racePair* - runs two *Task*s in parallel and returns when the first is complete. - returns an *Either* containing two tuples each containing a result value and a *Fiber* of the other *Task*. - If the 1st *Task* completes first the *Either* is a *Left* with the result of the first *Task* and a *Fiber* for the second *Task* (analogous if the *Either* is a *Right*). ```scala val random = scala.util.Random val task1 = Task(10 + 1).delayExecution(random.nextInt(3) seconds) val task2 = Task(20 + 1).delayExecution(random.nextInt(3) seconds) val raceTask: Task[Either[(Int, Fiber[Int]), (Fiber[Int], Int)]] = Task.racePair(task1, task2) implicit val scheduler: Scheduler = Scheduler.global val raceFuture: CancelableFuture[Either[(Int, Fiber[Int]), (Fiber[Int], Int)]] = raceTask.runToFuture raceFuture.foreach { case Left((result1, fiber)) => fiber.cancel println(s"Task 1 succeeded with result: $result1") case Right((fiber, result2)) => fiber.cancel println(s"Task 2 succeeded with result: $result2") } ``` --- ## *Task.race* - runs two *Task*s in parallel and returns after the first one completes. - The loser is cancelled automatically. - The resulting *Either* is a *Left* if the first one completes first, otherwise a *Right*. ```scala val random = scala.util.Random val task1 = Task(10 + 1).delayExecution(random.nextInt(3) seconds) val task2 = Task(20 + 1).delayExecution(random.nextInt(3) seconds) val winnerTask: Task[Either[Int, Int]] = Task.race(task1, task2) implicit val scheduler: Scheduler = Scheduler.global winnerTask .runToFuture .foreach(result => println(s"Winner's result: $result")) ``` --- ## *Task.raceMany* - runs a *Seq[Task[A]]* in parallel and returns the first that completes. - All loosers are cancelled automatically. ```scala val random = scala.util.Random val task1 = Task(10 + 1).delayExecution(random.nextInt(3) seconds) val task2 = Task(20 + 1).delayExecution(random.nextInt(3) seconds) val winnerTask: Task[Int] = Task.raceMany(Seq(task1, task2)) implicit val scheduler: Scheduler = Scheduler.global winnerTask .runToFuture .foreach(result => println(s"Winner's result: $result")) ``` --- class: middle, center name: DelayingTask # 16. Delaying a *Task*
--- ## Delaying a *Task* - Overview - For a *Task* you can delay the execution or the signaling of the result - either for a certain time duration or until another *Task* is complete. ```scala // delays the execution of a Task // for a certain time duration val delayed: Task[Int] = task.delayExecution(3.seconds) ``` ```scala // delays the execution of a Task // until another Task (the trigger) is complete val trigger: Task[Unit] = ??? val delayed: Task[Int] = trigger.flatMap(_ => source) ``` ```scala // delays signaling of a Task's result // for a certain time duration val delayed: Task[Int] = task.delayResult(3.seconds) ``` ```scala // delays signaling of a Task's result // until another Task (the selector) is complete def selector(x: Int): Task[Int] = ??? val delayed: Task[Int] = task.flatMap(x => selector(x)) ``` --- ## *Task#delayExecution*
- delays the execution of a *Task* for a certain time duration. ```scala val source = Task { println(s"side effect on $currentThread") "Hello" } val delayed: Task[String] = source .delayExecution(2.seconds) implicit val scheduler: Scheduler = Scheduler.global delayed.runToFuture foreach println ``` --- ## Delay a *Task* until another *Task* is complete
- delays the execution of a *Task* until another *Task* (the trigger) is complete. ```scala val source = Task { println(s"side effect on $currentThread") "Hello" } val delayed: Task[String] = trigger // trigger delays execution of source. .flatMap(_ => source) implicit val scheduler: Scheduler = Scheduler.global delayed.runToFuture foreach println ``` -- - alternatively with *> (= cats.Apply.productR): ```scala import cats.syntax.apply._ val delayed: Task[String] = trigger *> source ``` --- ## *Task#delayResult*
- delays signaling of a *Task*'s result for a certain time duration. ```scala val source = Task { println(s"side effect on $currentThread") "Hello" } val delayed: Task[String] = source .delayExecution(2.second) .delayResult(3.seconds) implicit val scheduler: Scheduler = Scheduler.global delayed.runToFuture foreach println ``` --- ## Delay signaling of a *Task*'s result until another *Task* is complete
- delays signaling of a *Task*'s result until another *Task* (the selector) is complete. ```scala val source = Task { println(s"side effect on $currentThread") Random.nextInt(5) } def selector(x: Int): Task[Int] = Task(x).delayExecution(x.seconds) val delayed: Task[Int] = source .delayExecution(1.second) .flatMap(x => selector(x)) // selector delays result signaling of source. implicit val scheduler: Scheduler = Scheduler.global delayed.runToFuture foreach { x => println(s"Result: $x (signaled after at least ${x+1} seconds)") } ``` --- class: middle, center name: Parallelism # 17. Parallelism (cats.Parallel)
--- ## The problem with *flatMap* - The three *Task* operations are independent of each other. - But *Task#flatMap* (for-comprehension) forces them to be sequential, (as if they were dependent). -- ```scala def sumTask(from: Int, to: Int): Task[Int] = Task { sumOfRange(from, to) } def factorialTask(n: Int): Task[BigInt] = Task { factorial(n) } def fibonacciTask(cycles: Int): Task[BigInt] = Task { fibonacci(cycles) } ``` ```scala val aggregate: Task[BigInt] = for { // sequential operations based on flatMap sum <- sumTask(0, 1000) fac <- factorialTask(10) fib <- fibonacciTask(10) } yield sum + fac + fib ``` ```scala implicit val scheduler: Scheduler = Scheduler.global aggregate runAsync printCallback def printCallback[A]: Callback[Throwable, A] = new Callback[Throwable, A] { def onSuccess(result: A): Unit = println(s"result = $result") def onError(ex: Throwable): Unit = println(s"ERROR: ${ex.toString}") } ``` --- ## The problem with *flatMap* - Composition with *flatMap* (often implicitly in for-comprehensions) enforces sequential, i.e. dependent execution. - *flatMap* prevents independent / parallel execution. - Type class *cats.Parallel* allows to switch between sequential (monadic) and parallel (applicative) execution. ```scala type ~>[F[_], G[_]] = arrow.FunctionK[F, G] trait Parallel[M[_], F[_]] { // Natural Transformation from the parallel Applicative F[_] to the sequential Monad M[_] def sequential: F ~> M // Natural Transformation from the sequential Monad M[_] to the parallel Applicative F[_]. def parallel: M ~> F } ``` - Monix *Task* has instance of *cats.Parallel*. - But it also provides its own (more efficient) impl of *parZip[n]*, *parMap[n]*, *parSequnce*, *parTraverse*. --- ## Replace *flatMap* / for-comprehension ... -- - with *Task#parZip3* followed by *map* ```scala val aggregate: Task[BigInt] = // potentially executed in parallel Task.parZip3(sumTask(0, 1000), factorialTask(10), fibonacciTask(10)).map { (sum, fac, fib) => sum + fac + fib } ``` -- - with *Task#parMap3* ```scala val aggregate: Task[BigInt] = // potentially executed in parallel Task.parMap3(sumTask(0, 1000), factorialTask(10), fibonacciTask(10)) { (sum, fac, fib) => sum + fac + fib } ``` -- - with *parMapN* using *cats.syntax.parallel._* ```scala import cats.syntax.parallel._ val aggregate: Task[BigInt] = // potentially executed in parallel (sumTask(0, 1000), factorialTask(10), fibonacciTask(10)) parMapN { _ + _ + _ } ``` --- ## Sequencing - Overview *Task* provides three variants for sequencing (analogous to *Future.sequence*):
-- - *Task#sequence* ```scala // sequentially converts a *Seq[Task[A]]* to *Task[Seq[A]]* // the order of side effects is guaranteed as well as the order of results val taskOfSeq: Task[Seq[Int]] = Task.sequence(seqOfTask) ``` -- - *Task#gather* ```scala // converts a *Seq[Task[A]]* to *Task[Seq[A]]* in parallel // the order of side effects is not guaranteed but the order of results is val taskOfSeq: Task[Seq[Int]] = Task.gather(seqOfTask) ``` -- - *Task#gatherUnordered* ```scala // converts a *Seq[Task[A]]* to *Task[Seq[A]]* in parallel // the order of side effects is not guaranteed, neither is the order of results val taskOfSeq: Task[Seq[Int]] = Task.gatherUnordered(seqOfTask) ``` --- ## *Task#sequence*
- sequentially converts a *Seq[Task[A]]* to *Task[Seq[A]]*. - The order of side effects is guaranteed as well as the order of results. ```scala val task1 = Task { println("side effect 1"); 1 } val task2 = Task { println("side effect 2"); 2 } val seqOfTask: Seq[Task[Int]] = Seq(task1, task2) val taskOfSeq: Task[Seq[Int]] = Task.sequence(seqOfTask) implicit val scheduler: Scheduler = Scheduler.global // We always get the same ordering in the output: taskOfSeq foreach println //=> side effect 1 //=> side effect 2 //=> List(1, 2) taskOfSeq foreach println //=> side effect 1 //=> side effect 2 //=> List(1, 2) ``` --- ## *Task#gather*
- converts a *Seq[Task[A]]* to *Task[Seq[A]]* in parallel. - The order of side effects is not guaranteed but the order of results is. ```scala val task1 = Task { println("side effect 1"); 1 }.delayExecution(1 second) val task2 = Task { println("side effect 2"); 2 }.delayExecution(1 second) val seqOfTask: Seq[Task[Int]] = Seq(task1, task2) val taskOfSeq: Task[Seq[Int]] = Task.gather(seqOfTask) implicit val scheduler: Scheduler = Scheduler.global // There’s potential for parallel execution: // Ordering of effects is not guaranteed, but the results in the List are ordered. taskOfSeq foreach println //=> side effect 1 //=> side effect 2 //=> List(1, 2) taskOfSeq foreach println //=> side effect 2 //=> side effect 1 //=> List(1, 2) ``` --- ## *Task#gatherUnordered*
- converts a *Seq[Task[A]]* to *Task[Seq[A]]* in parallel. - The order of side effects is not guaranteed, neither is the order of results. ```scala val task1 = Task { println("side effect 1"); 1 }.delayExecution(1 second) val task2 = Task { println("side effect 2"); 2 }.delayExecution(1 second) val seqOfTask: Seq[Task[Int]] = Seq(task1, task2) val taskOfSeq: Task[Seq[Int]] = Task.gatherUnordered(seqOfTask) implicit val scheduler: Scheduler = Scheduler.global // There’s potential for parallel execution: // Ordering of effects is not guaranteed, results in the List are unordered too. taskOfSeq foreach println //=> side effect 1 //=> side effect 2 //=> List(1, 2) taskOfSeq foreach println //=> side effect 2 //=> side effect 1 //=> List(2, 1) ``` --- ## Traversing - Overview *Task* provides three variants for traversing (analogous to *Future.traverse*):
-- - *Task#traverse* ```scala // sequentially traverses a *Seq[A]* with a function A => Task[A] to *Task[Seq[A]]* // the order of side effects is guaranteed as well as the order of results val taskOfSeq: Task[Seq[Int]] = Task.traverse(Seq(1, 2))(i => task(i)) ``` -- - *Task#wander* ```scala // traverses a *Seq[A]* with a function A => Task[A] to *Task[Seq[A]]* in parallel // the order of side effects is not guaranteed but the order of results is val taskOfSeq: Task[Seq[Int]] = Task.wander(Seq(1, 2))(i => task(i)) ``` -- - *Task#wanderUnordered* ```scala // traverses a *Seq[A]* with a function A => Task[A] to *Task[Seq[A]]* in parallel // the order of side effects is not guaranteed, neither is the order of results val taskOfSeq: Task[Seq[Int]] = Task.wanderUnordered(Seq(1, 2))(i => task(i)) ``` --- ## *Task#traverse*
- sequentially traverses a *Seq[A]* with a function A => Task[A] to *Task[Seq[A]]*. - The order of side effects is guaranteed as well as the order of results. ```scala def task(i: Int): Task[Int] = Task { println("side effect " + i); i } val taskOfSeq: Task[Seq[Int]] = Task.traverse(Seq(1, 2))(i => task(i)) implicit val scheduler: Scheduler = Scheduler.global // We always get the same ordering in the output: taskOfSeq foreach println //=> side effect 1 //=> side effect 2 //=> List(1, 2) taskOfSeq foreach println //=> side effect 1 //=> side effect 2 //=> List(1, 2) ``` --- ## *Task#wander*
- traverses a *Seq[A]* with a function A => Task[A] to *Task[Seq[A]]* in parallel. - The order of side effects is not guaranteed but the order of results is. ```scala def task(i: Int): Task[Int] = Task { println("side effect " + i); i }.delayExecution(1 second) val taskOfSeq: Task[Seq[Int]] = Task.wander(Seq(1, 2))(i => task(i)) implicit val scheduler: Scheduler = Scheduler.global // There’s potential for parallel execution: // Ordering of effects is not guaranteed, but the results in the List are ordered. // list.runToFuture.foreach(println) taskOfSeq foreach println //=> side effect 1 //=> side effect 2 //=> List(1, 2) // list.runToFuture.foreach(println) taskOfSeq foreach println //=> side effect 2 //=> side effect 1 //=> List(1, 2) ``` --- ## *Task#wanderUnordered*
- traverses a *Seq[A]* with a function A => Task[A] to *Task[Seq[A]]* in parallel. - The order of side effects is not guaranteed, neither is the order of results. ```scala def task(i: Int): Task[Int] = Task { println("side effect " + i); i }.delayExecution(1 second) val taskOfSeq: Task[Seq[Int]] = Task.wanderUnordered(Seq(1, 2))(i => task(i)) implicit val scheduler: Scheduler = Scheduler.global // There’s potential for parallel execution: // Ordering of effects is not guaranteed, results in the List are unordered too. taskOfSeq foreach println //=> side effect 1 //=> side effect 2 //=> List(1, 2) taskOfSeq foreach println //=> side effect 2 //=> side effect 1 //=> List(2, 1) ``` --- class: middle, center name: TaskApp # 18. TaskApp
--- ## *TaskApp* - allows you to create an App where you only define a *Task* - You define the *Task* inside *TaskApp#run*. - *TaskApp#run* takes the command line args as a *List[String]* and returns a *Task[ExitCode]*. - No need to invoke one of the run* methods like *runAsync*, *runToFuture* or *foreach* - No need to provide an implicit *Scheduler*. -- ```scala object MyApp extends TaskApp { def run(args: List[String]): Task[ExitCode] = args.headOption match { case Some(name) => Task(println(s"Hello, $name.")).as(ExitCode.Success) case None => Task(System.err.println("Usage: MyApp name")).as(ExitCode(2)) } } ``` --- ## *TaskApp* - Another example ```scala object FibonacciApp extends TaskApp { def run(args: List[String]): Task[ExitCode] = args.headOption match { case None => Task(System.err.println("Usage: MyApp cycles")).as(ExitCode(1)) case Some(cyclesStr) => Try(cyclesStr.toInt).toEither match { case Left(t) => Task(System.err.println("Not a positive number")).as(ExitCode(2)) case Right(cycles) => defineWork(cycles).as(ExitCode.Success) } } def defineWork(cycles: Int): Task[Unit] = for { result <- Task { fibonacci(cycles) } _ <- Task(println("\n-----")) _ <- Task(println(s"Fibonacci for $cycles cycles is: $result.")) _ <- Task(println("-----\n")) } yield () } ``` --- class: middle, center name: CompletableFuture # 19. CompletableFuture
--- ## *FutureUtils*
- is an object with utility functions for *scala.concurrent.Future*. - Provides (amongst other things) interop with Java's *CompletableFuture* - *fromJavaCompletable* converts a *CompletableFuture* to a Scala *Future*. - *toJavaCompletable* converts a Scala *Future* to a *CompletableFuture*. --- ## *FutureUtils.fromJavaCompletable*
- converts a *CompletableFuture* to a *scala.concurrent.Future* ```scala implicit val ec: ExecutionContextExecutor = ExecutionContext.global val supplier: Supplier[BigInt] = () => fibonacci(6) def completable: CompletableFuture[BigInt] = CompletableFuture.supplyAsync(supplier) val future: Future[BigInt] = FutureUtils.fromJavaCompletable(completable) future foreach println ``` --- ## Converting *CompletableFuture* to *Task*
```scala def completable: CompletableFuture[BigInt] = CompletableFuture.supplyAsync(() => fibonacci(6)) def future: Future[BigInt] = FutureUtils.fromJavaCompletable(completable) val task = Task.deferFutureAction { implicit scheduler => future } implicit val scheduler: Scheduler = Scheduler.global task foreach println ``` --- ## *FutureUtils.toJavaCompletable*
- converts a *scala.concurrent.Future* to a *CompletableFuture* ```scala implicit val ec: ExecutionContext = ExecutionContext.global val future: Future[BigInt] = Future { fibonacci(6) } val completable: CompletableFuture[BigInt] = FutureUtils.toJavaCompletable(future) val consumer: Consumer[BigInt] = new Consumer[BigInt] { override def accept(value: BigInt): Unit = println(value) } completable.thenAccept(consumer) ``` --- ## Converting *Task* to *CompletableFuture*
```scala val task: Task[BigInt] = Task { fibonacci(6) } implicit val scheduler: Scheduler = Scheduler.global val completable: CompletableFuture[BigInt] = FutureUtils.toJavaCompletable(task.runToFuture) completable.thenAccept((value: BigInt) => println(value)) ``` --- class: middle, center name: ResourceManagement # 20. Resources Management
--- ## Line count example with classic Java File IO
```scala def doCountLines(in: BufferedReader): Int = { var count: Int = 0 var line: String = in.readLine() while (line != null) { count += 1 line = in.readLine() } count } def countLines(file: File): Task[Int] = Task { val in = new BufferedReader(new FileReader(file)) try { doCountLines(in) } finally { in.close() } } val task: Task[Int] = countLines(new File("README.md")) implicit val scheduler: Scheduler = Scheduler.global task foreach println ``` --- ## *Task#bracket*
- can separate resource acquisition, it's usage and it's release ```scala def doCountLines(in: BufferedReader): Int = ??? // as before def countLines(file: File): Task[Int] = { val acquire: Task[BufferedReader] = Task { new BufferedReader(new FileReader(file)) } acquire.bracket { in => Task { doCountLines(in) } } { in => Task { in.close() } } } val task: Task[Int] = countLines(new File("README.md")) implicit val scheduler: Scheduler = Scheduler.global task foreach println ``` --- ## *cats.effect.Resource*
- *Resource.make* implements resource acquisition and release - provides a method *use* that take a function for resource usage ```scala def doCountLines(in: BufferedReader): Int = ??? // as before def openFile(file: File): Resource[Task, BufferedReader] = Resource.make( Task(new BufferedReader(new FileReader(file))) )( in => Task(in.close()) ) def countLines(file: File): Task[Int] = openFile(file).use { in => Task { doCountLines(in) } } val task: Task[Int] = countLines(new File("README.md")) implicit val scheduler: Scheduler = Scheduler.global task foreach println ``` --- ## *Resource.fromAutoCloseable*
- simplifies *Resource.make* by omitting the release function ```scala def doCountLines(in: BufferedReader): Int = ??? // as before def openFile(file: File): Resource[Task, BufferedReader] = Resource.fromAutoCloseable( Task(new BufferedReader(new FileReader(file))) ) def countLines(file: File): Task[Int] = openFile(file).use { in => Task { doCountLines(in) } } val task: Task[Int] = countLines(new File("README.md")) implicit val scheduler: Scheduler = Scheduler.global task foreach println ``` --- class: middle, center name: Resources # 21. Resources
--- ## Resources (1/2) - Code and Slides of this Talk:
https://github.com/hermannhueck/future-vs-monix-task - Code and Slides for my own Implementation of IO Monad:
https://github.com/hermannhueck/implementing-io-monad - Monix 3.x Documentation
https://monix.io/docs/3x/ - Monix 3.x API Documentation
https://monix.io/api/3.0/ - Monix Task: Lazy, Async and Awesome
Talk by Alexandru Nedelcu at Scala Days 2017
https://www.youtube.com/watch?v=wi97X8_JQUk - Converting Scala's Future to Task
Video Tutorial by Alexandru Nedelcu
https://monix.io/blog/2018/11/08/tutorial-future-to-task.html - Presentations on Monix:
https://monix.io/presentations --- ## Resources (2/2) - "Concurrency Basics" in cats-effect
https://typelevel.org/cats-effect/concurrency/basics.html - Daniels Spiewak's gist on Thread Pools
https://gist.github.com/djspiewak/46b543800958cf61af6efa8e072bfd5c - Daniels Spiewak's blog: An IO monad for cats
https://typelevel.org/blog/2017/05/02/io-monad-for-cats.html - Discussion about Green Threads and blocking I/O
https://github.com/typelevel/cats-effect/issues/243 - Best Practice: "Should Not Block Threads"
https://monix.io/docs/3x/best-practices/blocking.html - What Referential Transparency can do for you
Talk by Luka Jacobowitz at ScalaIO 2017
https://www.youtube.com/watch?v=X-cEGEJMx_4 --- class: middle, center ## Thanks for Listening
# Q & A
https://github.com/hermannhueck/future-vs-monix-task


the book of satoshi
The Collected Writings of Bitcoin Creator Satoshi Nakamoto
Learn about Bitcoin and its early history
Have you, like the rest of the world, speculated as to the identity of Satoshi Nakamoto, anonymous creator of Bitcoin?
The world’s first cryptocurrency, Bitcoin went online in 2009 and has since revolutionized our concepts of currency and money. Not supported by any government or central bank, completely electronic, Bitcoin is a virtual currency based on advanced cryptographic systems.
Like the currency he created, the identity of Bitcoin’s creator Satoshi Nakamoto is virtual, existing only online. The Nakamoto persona, which may represent an individual or a group, exists only in the online publications that introduced and explained Bitcoin during its earliest days. Here, collected and professionally published for the first time are the essential writings that detail Bitcoin’s creation. Included are
- Satoshi Nakamoto Emails and Posts on Computer Forums Presented in Chronological Order
- Bitcoin Fundamentals Presented in Layman’s Terms
- Bitcoin’s Potential and Profound Economic Implications
- The Seminal Paper Which Started It All
The Book of Satoshi provides a convenient way to parse through what Bitcoin’s creator wrote over the span of the two years that constituted his “public life" before he disappeared from the Internet . . . at least under the name Satoshi Nakamoto. Beginning on November 1st 2009 with the publication of the seminal paper describing Bitcoin, this public life ends at about the time PC World speculated as to a possible link between Bitcoin and WikiLeaks, the infamous website that publishes leaked classified materials. Was there a connection? You be the judge.
Nakamoto’s true identity may never be known. Therefore the writings reproduced here are probably all the world will ever hear from him concerning Bitcoin’s creation, workings, and theoretical basis. Want to learn more about Bitcoin? Go directly to the source—the writings of the creator himself, Satoshi Nakamoto!


NextDatabaseDriver()
Syntax
Result = NextDatabaseDriver()Description
Retrieves information about the next available database driver. This function must be called after ExamineDatabaseDrivers(). To get information about the current driver, DatabaseDriverName() and DatabaseDriverDescription() can be used.
Parameters
None.
Return value
If Result is 0, no more drivers are available.
Remarks
This is an ODBC database specific command.
See Also
ExamineDatabaseDrivers(), DatabaseDriverName(), DatabaseDriverDescription()
Supported OS
All


Bears fans trade punches, fight in stands during game (Video)
by TSD StaffThe Chicago Bears entered the season in hopes of defending their NFC North title, and advancing further in the playoffs than last year, but the team’s 2019 campaign has been filled with nothing but disappointment.
Chicago entered Sunday’s game against New York sporting a lackluster 4-6 record, and it’s’ fair to say fans at Soldier Field were hoping the team would be taking playoffs this time of year, rather than if quarterback Mitchell Trubisky is the long-term answer.
All of these things have taken a toll on fans, and is probably why Bears supporters were so frustrated that they began just fighting each other in the stands at Soldier Field on Sunday.
Yeesh.


TURN server installation Guide
HOME © Muaz Khan . @WebRTCWeb . Github . Latest issues . What's New?
This document is aimed to explain TURN server installation steps for different operating systems.
- CoTURN installation on Ubuntu
- restund installation on Ubuntu
- TURN installation on CentOS
- TURN installation on Windows
- reTurnServer installation on Ubuntu
- Pion TURN server
You can test a TURN here LIVE or use "apt-get install stun" and then "stun domain.com:port".
Top
CoTURN installation on Ubuntu
- Make sure that you're using latest up-to-dated Ubuntu (TLS 14+)
-
Find a useful FTP link for your region.
https://packages.debian.org/jessie/amd64/coturn/download
Simply copy the link. Don't try any command yet. -
Modify sources.list file:
vi /etc/apt/sources.list
And add above FTP_domain in the sources-list. E.g.
# at the bottom of the sources.list file # assuming that you found this domain: # http://ftp.hk.debian.org/debian deb http://ftp.hk.debian.org/debian jessie main
-
Make sure you have installed "Aptitude":
# https://wiki.debian.org/Aptitude # update-to-latest version aptitude update
-
If "aptitude update" fails:
# replace "8B48AD6246925553" and "7638D0442B90D010" with the error_key displayed on your OWN-screen # GPG error: http://ftp.hk.debian.org jessie Release: # The following signatures could not be verified because the public key is # not available: NO_PUBKEY 8B48AD6246925553 NO_PUBKEY 7638D0442B90D010 NO_PUBKEY CBF8D6FD518E17E1 # execute below commands gpg --keyserver pgpkeys.mit.edu --recv-key 8B48AD6246925553 gpg -a --export 8B48AD6246925553 | apt-key add - gpg --keyserver pgpkeys.mit.edu --recv-key 7638D0442B90D010 gpg -a --export 7638D0442B90D010 | apt-key add -
-
Try "Aptitude" again if failed on initial try:
aptitude update
-
Now install "coturn" package using "aptitude":
# below command may require [sudo] privileges aptitude install coturn
-
Modify following file after "coturn" installation:
vi /etc/turnserver.conf
Remove all text from the file, and replace with this:
# you can listen ports 3478 and 5349 instead of 80/443 listening-port=80 tls-listening-port=443 listening-ip=your-ip-address relay-ip=your-ip-address external-ip=your-ip-address realm=yourdomain.com server-name=yourdomain.com lt-cred-mech userdb=/etc/turnuserdb.conf # use real-valid certificate/privatekey files cert=/etc/ssl/certificate.pem pkey=/etc/ssl/private.key no-stdout-log
-
Modify following file after modifying above "turnserver.conf":
vi /etc/turnuserdb.conf
Remove all text from the file, and replace with this:
yourName:yourPassword
-
Instead of "turnuserdb.conf", you can try following as well:
# rquires: apt-get install sqlite3 libsqlite3-dev # now: find / -name 'turndb' # vi /etc/turnserver.conf userdb=/var/lib/turn/turndb # in this case, please add following: # lt-cred-mech # remove or comment this one oauth # add this user=youruser:yourpassword # add this # source: askubuntu.com/a/819264
-
Now type following command:
turnserver
-
Try following if above "turnserver" command fails:
# you can listen ports 3478 and 5349 instead of 80/443 lsof -n -i4TCP:80 | grep LISTEN lsof -n -i4TCP:443 | grep LISTEN
Kill "existing" processes using this command:
kill processId
-
Now type following command again and it should work:
turnserver
-
Make sure that ports 80/443 (or 3478/5349) are opened in the firewall:
# you can listen ports 3478 and 5349 instead of 80/443 iptables -A INPUT -p udp --dport 80 -j ACCEPT iptables -A INPUT -p udp --dport 443 -j ACCEPT
-
Run following command:
# pm2 is strongly recommended # https://github.com/Unitech/pm2 pm2 start turnserver pm2 save # after: pm2 startup # if you do not want to use "pm2" nohup turnserver > /dev/null 2>&1 & # or simply nohup turnserver &
-
Now check all UDP+TCP ports:
# recommended netstat -tulpn # or netstat -nat | grep LISTEN
You will see that port 80/443 is listened by turnserver (both for UDP/TCP). -
Otherwise (alternative coTURN installation solutions):
// Please follow this link: groups.google.com/d/msg/easyrtc/ypjJ5Yu3wZM/u5Lq6VNfabcJ // or apt-get install coturn // or: stackoverflow.com/a/37290864 sudo -i # ignore if you already in admin mode apt-get update && apt-get install libssl-dev libevent-dev libhiredis-dev make -y # install the dependencies wget -O turn.tar.gz http://turnserver.open-sys.org/downloads/v<IP_ADDRESS>/turnserver-<IP_ADDRESS>.tar.gz # Download the source tar tar -zxvf turn.tar.gz # unzip cd turnserver-* ./configure make && make install // or: stackoverflow.com/a/52408246
Top
restund installation on Ubuntu (via)
"make" and "gcc" are required. Example LIVE server
-
Download "re":
mkdir /home/restund cd /home/restund # via: http://creytiv.com/pub/ wget http://webrtcweb.com/re-latest.tar.gz tar zxvf re-latest.tar.gz cd re-0.4.17 make make install # "sudo" is preferred
-
Download "restund":
# via: http://www.creytiv.com/pub/ # choose latest version # replace "0.4.12" with latesat version (06-Oct-2015) cd .. wget http://webrtcweb.com/restund-0.4.12.tar.gz tar zxvf restund-0.4.12.tar.gz cd restund-0.4.12 make make install # "sudo" is preferred
-
Modify "etc/restund.conf" file:
It will be located at: restund-0.4.12/etc/restund.conf
# cd etc && vi restund.conf # "etc" is a directory inside "restund-0.4.12" udp_listen your-ip:4455 tcp_listen your-ip:5544 # different port than UDP turn_relay_addr your-ip debug yes realm yourdomain.com # you will get a HTTP server for stats # example stats: # Version: 0.4.12 # Built: Nov 10 2016 23:39:41 # Uptime: 21 hours 25 mins 56 secs # LIVE Example: http://webrtcweb.com:4050/ status_udp_addr your-ip status_udp_port 33000 status_http_addr your-ip status_http_port 4050 # filedb (your username/password are stored in this file) filedb_path /etc/restund.auth # modules (STUN messages are processed in module loading order) module_path /usr/local/lib/restund/modules module stat.so module binding.so module auth.so module turn.so module filedb.so module syslog.so module status.so syncinterval 600 auth_nonce_expiry 3600
-
Modify "etc/restund.auth" file:
It will be located at: restund-0.4.12/etc/restund.auth
# remove default credentials
First of all, execute this command:
cd .. util/genha1.sh username yourdomain.com password
Above command will print something. Copy whatever is printed; and paste inside "etc/restund.auth":
cd etc vi restund.auth # copy/paste inside the "restund.auth" file username:sha1
-
Copy all "etc" files to "/etc/"
cd .. # move .conf and .auth files to /etc/ directory cp ./etc/* /etc/ # or use absolute path: # cp /home/restund/restund-0.4.12/etc/ /etc/
-
Run the "restund" server
restund
Above command will print something like this:
udp listen: your-ip:4455 tcp listen: your-ip:5544 stat: module loaded binding: module loaded auth: module loaded (nonce_expiry=3600s) turn: lifetime=600 ext=your-ip ext6=::1 bsz=512 filedb: module loaded (/etc/restund.auth) syslog: module loaded facility=24 status: module loaded (udp=your-ip:33000 http=your-ip:4050)
You can open stats server here: http://your-ip:4050
-
If webrtc access to "restund" fails:
iptables -I INPUT 1 -i eth0 -p udp --dport 33000 -j ACCEPT iptables -I INPUT 1 -i eth0 -p udp --dport 4455 -j ACCEPT iptables -I INPUT 1 -i eth0 -p tcp --dport 4455 -j ACCEPT iptables -I INPUT 1 -i eth0 -p udp --dport 5544 -j ACCEPT iptables -I INPUT 1 -i eth0 -p tcp --dport 5544 -j ACCEPT iptables -I INPUT 1 -i eth0 -p tcp --dport 4050 -j ACCEPT
-
Now open this: http://your-ip.com:4050/
Top
TURN installation on CentOS
Use "sudo" if you see "permission-denied" errors.
-
First Step: Download & Install Prerequisite for CentOS
yum install -y make gcc cc gcc-c++ wget yum install -y openssl-devel libevent libevent-devel mysql-devel mysql-server
-
Second Step: Download & Install LibEvent modules
wget https://github.com/downloads/libevent/libevent/libevent-2.0.21-stable.tar.gz tar xvfz libevent-2.0.21-stable.tar.gz cd libevent-2.0.21-stable && ./configure make && make install && cd .. # you can use "sudo" if permission-denied kind of erros occurred. # sudo make && sudo make install && cd ..
-
Third Step: Download & Install TURN modules
wget http://turnserver.open-sys.org/downloads/v<IP_ADDRESS>/turnserver-<IP_ADDRESS>.tar.gz tar -xvzf turnserver-<IP_ADDRESS>.tar.gz cd turnserver-<IP_ADDRESS> && ./configure make && make install
-
Fifth Step: Configure "turnserver.conf" file
You can:vi /etc/turnserver/turnserver.conf # setting static accounts # Remember, "static" accounts are not dynamically checked by the turnserver process. user=username:password # listen ports listening-port=2222 listening-ip=127.1.1 # Now press "insert" key; then "Esc" key # Then type: :wq #-- this command will save your settings and close turnserver.conf file # if you don't want to save settings; and quite. Simply type: :q
- Use Only STUN
- Use Only TURN
- Use Both STUN and TURN
- Discard requests from all SSL/TLS version of protocols
- Use short-live credentials. Other authentication mechanisms coming soon in next few weeks.
- You can disable DTLS and TLS.
- You can set both static and dynamic accounts.
- You can give anonymous access to TURN server as well.
- You can set credentials for stun-only option as well; usually STUN-binding requests are anonymous.
-
Sixth Step: Run TURN server instance
cd /etc/turnserver/ turnserver -v -r ip:port -a -b turnuserdb.conf -c turnserver.conf -u turn-username -r ip:port -p turn-password
- Replace "ip:port" with yours! E.g. 127.1.1:2222:
-
turnserver -v -r 127.1.1:2222 -a -b turnuserdb.conf -c turnserver.conf -u turn-username -r 127.1.1:2222 -p turn-password
- Replace "turn-username" with custom-username; and "turn-password" with custom-password.
-
turnserver -v -r 127.1.1:2222 -a -b turnuserdb.conf -c turnserver.conf -u custom-username -r 127.1.1:2222 -p custom-password
-
Last Step: "Permanently" run TURN server instance
nohup turnserver -v -r ip:port -a -b turnuserdb.conf -c turnserver.conf -u turn-username -r ip:port -p turn-password &
- You can see that the entire TURN execution command is placed between "nohup" and "&".
-
nohup TURN-execution-command &
Top
TURN installation on Windows
Alternatives: STUNTMAN or C# STUN Client or Pion TURN-server for Windows.
Please check PION link above for a Windows TURN client.
Note: Below section is taken from slideshare; and its credit goes to @amiteshawa!
- Install "Cygwin" application
- Follow all steps from above section. E.g. install all dependencies; LibEvent and TURN modules.
-
You'll face some issues when invoking "./configure" command for TURN configuration:
- Missing libpq.a, hiredis, postgreSql
- You can ignore them, because these modules are optional
- During "make" command invocation; you'll face error like "error: 'struct sockaddr_in' has no member named 'sin_len'"
-
Edit "na_turn_ioaddr.c" and comment line number "169":
vi /turnserver-<IP_ADDRESS>/src/client/na_turn_ioaddr.c # commenting line number #169 # addr->s4.sin_len = sizeof(struct sockaddr_in);
- Now, "make" and "make install" commands for TURN modules installation will work.
- Then copy "turnserver.conf" file from "../examples/etc/" directory to "/usr/local/etc/":
-
cp /turnserver-<IP_ADDRESS>/examples/etc/turnserver.conf /usr/local/etc/
- Now, you can follow "fifth" step from previous section to edit "turnserver.conf" for setting "username", "password", "listening-port" and "listening-ip".
Top
reTurnServer installation on Ubuntu
-
First Step: Install reTurnServer:
# Ubuntu apt-get install resiprocate-turn-server # CentOS yum install resiprocate-turn-server
-
Second Step: Edit "/etc/reTurn/reTurnServer.config" file:
TurnAddress = your-first-ip AltStunAddress = your-second-ip TurnPort = 3478 AltStunPort = 5349 AuthenticationRealm = domain.com UserDatabaseFile = /etc/reTurn/users.txt UserDatabaseHashedPasswords = true # required for HashString # for more info: # https://github.com/resiprocate/resiprocate/blob/master/reTurn/reTurnServer.config
-
Next Step: Edit "/etc/reTurn/users.txt" file:
# first of all, execute this command: # echo -n username:domain.com:password | md5sum # you will get a HASH string: 32 chars # domain.com must match "AuthenticationRealm" above "reTurnServer.config" username:AboveHashString:domain.com:authorized # more info: # https://github.com/resiprocate/resiprocate/blob/master/reTurn/users.txt
-
Last Step: ReStart reTurnServer:
service resiprocate-turn-server restart
-
You can check ports and addresses reTurnServer is listenning on:
netstat -nlp | grep reTurnServer
-
In your HTML-JavaScript file, you will use original password, NOT the HashString:
iceServer => { urls: 'turn:domain.com:3478', credential: 'orignal-password', // NOT Hash username: 'username' }
Top
Pion TURN server
A simple extendable Golang TURN server for Windows, Linux, Darwin and FreeBSD.
-
Create a new directory (optional):
mkdir pions cd pions
-
Download the TURN server's source: (replace "1.0.3" with latest release)
wget -q https://github.com/pions/turn/releases/download/1.0.3/simple-turn-linux-amd64 # linux: simple-turn-linux-386 or simple-turn-linux-amd64 # darwin: simple-turn-darwin-amd64 # freebsd: simple-turn-freebsd-386 and simple-turn-freebsd-amd64 # windows: simple-turn-windows-386.exe and simple-turn-windows-amd64.exe
-
Give read-write permissions to the DIR:
chmod +x simple-turn-linux-amd64
-
Set username, password and port: (using EXPORT commands)
export USERS='username=password' export REALM=domain.com export UDP_PORT=3478
-
Now run the pions turn server:
./simple-turn-linux-amd64 # or use nohup nohup ./simple-turn-linux-amd64 &
- To check latest releases: https://github.com/pions/turn/releases