Skip to content

Commit 5a63d21

Browse files
simerplahasimerplaha
committed
#127 - Monix support
1 parent ee7d5bc commit 5a63d21

4 files changed

Lines changed: 271 additions & 57 deletions

File tree

‎build.sbt‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ val scalaMockVersion = "4.4.0"
1111
val scalaTestVersion = "3.0.8"
1212
val reactiveStreamsVersion = "1.0.2"
1313
val boopickleVersion = "1.3.1"
14+
val monixVersion = "3.0.0"
1415

1516
parallelExecution in ThisBuild := false
1617

@@ -68,7 +69,7 @@ lazy val SwayDB =
6869
.settings(commonSettings)
6970
.settings(publishSettings)
7071
.dependsOn(swaydb)
71-
.aggregate(swaydb, core, compression, data, configs, serializers)
72+
.aggregate(swaydb, core, compression, data, configs, serializers, monix)
7273

7374
lazy val core =
7475
project
@@ -111,6 +112,15 @@ lazy val serializers =
111112
.settings(publishSettings)
112113
.dependsOn(data)
113114

115+
lazy val monix =
116+
project
117+
.settings(commonSettings)
118+
.settings(publishSettings)
119+
.settings(
120+
libraryDependencies += "io.monix" %% "monix" % monixVersion
121+
)
122+
.dependsOn(data)
123+
114124
lazy val `core-stress` =
115125
project
116126
.settings(commonSettings)

‎data/src/main/scala/swaydb/Monad.scala‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
/*
2+
* Copyright (c) 2019 Simer Plaha (@simerplaha)
3+
*
4+
* This file is a part of SwayDB.
5+
*
6+
* SwayDB is free software: you can redistribute it and/or modify
7+
* it under the terms of the GNU Affero General Public License as
8+
* published by the Free Software Foundation, either version 3 of the
9+
* License, or (at your option) any later version.
10+
*
11+
* SwayDB is distributed in the hope that it will be useful,
12+
* but WITHOUT ANY WARRANTY; without even the implied warranty of
13+
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14+
* GNU Affero General Public License for more details.
15+
*
16+
* You should have received a copy of the GNU Affero General Public License
17+
* along with SwayDB. If not, see <https://www.gnu.org/licenses/>.
18+
*/
19+
20+
package swaydb
21+
22+
import scala.concurrent.{ExecutionContext, Future}
23+
24+
trait Monad[T[_]] {
25+
def map[A, B](a: A, f: A => B): T[B]
26+
def flatMap[A, B](a: T[A], f: A => T[B]): T[B]
27+
def success[A](a: A): T[A]
28+
def failed[A](a: Throwable): T[A]
29+
}
30+
31+
object Monad {
32+
implicit class Map[A](value: A) {
33+
@inline def map[B, T[_]](f: A => B)(implicit monad: Monad[T]): T[B] =
34+
monad.map(value, f)
35+
}
36+
37+
implicit class FlatMap[A, T[_]](value: T[A])(implicit monad: Monad[T]) {
38+
@inline def flatMap[B](f: A => T[B]): T[B] =
39+
monad.flatMap(value, f)
40+
}
41+
42+
implicit def futureMonad(implicit ec: ExecutionContext): Monad[Future] =
43+
new Monad[Future] {
44+
override def map[A, B](a: A, f: A => B): Future[B] =
45+
Future(f(a))
46+
47+
override def flatMap[A, B](a: Future[A], f: A => Future[B]): Future[B] =
48+
a.flatMap(f)
49+
50+
override def success[A](a: A): Future[A] =
51+
Future.successful(a)
52+
53+
override def failed[A](a: Throwable): Future[A] =
54+
Future.failed(a)
55+
}
56+
}

‎data/src/main/scala/swaydb/Tag.scala‎

Lines changed: 83 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -293,6 +293,75 @@ object Tag extends LazyLogging {
293293
}
294294
}
295295

296+
object Async {
297+
298+
import Monad._
299+
300+
def foldLeft[A, U, T[_]](initial: U, after: Option[A], stream: swaydb.Stream[A, T], drop: Int, take: Option[Int], operation: (U, A) => U)(implicit monad: Monad[T]): T[U] = {
301+
def fold(previous: A, drop: Int, currentSize: Int, previousResult: U): T[U] =
302+
if (take.contains(currentSize))
303+
monad.success(previousResult)
304+
else
305+
stream
306+
.next(previous)
307+
.flatMap {
308+
case Some(next: A) =>
309+
if (drop >= 1) {
310+
fold(next, drop - 1, currentSize, previousResult)
311+
} else {
312+
try {
313+
val newResult = operation(previousResult, next)
314+
fold(next, drop, currentSize + 1, newResult)
315+
} catch {
316+
case throwable: Throwable =>
317+
monad.failed(throwable)
318+
}
319+
}
320+
321+
case None =>
322+
monad.success(previousResult)
323+
}
324+
325+
if (take.contains(0))
326+
monad.success(initial)
327+
else
328+
after
329+
.map(stream.next)
330+
.getOrElse(stream.headOption)
331+
.flatMap {
332+
case Some(first) =>
333+
if (drop >= 1) {
334+
fold(first, drop - 1, 0, initial)
335+
} else {
336+
try {
337+
val nextResult = operation(initial, first)
338+
fold(first, drop, 1, nextResult)
339+
} catch {
340+
case throwable: Throwable =>
341+
monad.failed(throwable)
342+
}
343+
}
344+
345+
case None =>
346+
monad.success(initial)
347+
}
348+
}
349+
350+
def collectFirst[A, T[_]](previous: A, stream: swaydb.Stream[A, T], condition: A => Boolean)(implicit monad: Monad[T]): T[Option[A]] =
351+
stream
352+
.next(previous)
353+
.flatMap {
354+
case some @ Some(nextA) =>
355+
if (condition(nextA))
356+
monad.success(some)
357+
else
358+
collectFirst(nextA, stream, condition)
359+
360+
case None =>
361+
monad.success(None)
362+
}
363+
}
364+
296365
implicit val throwableIO: Tag.Sync[IO.ThrowableIO] =
297366
new Tag.Sync[IO.ThrowableIO] {
298367
override val unit: IO.ThrowableIO[Unit] =
@@ -465,64 +534,22 @@ object Tag extends LazyLogging {
465534
def isComplete[A](a: Future[A]): Boolean =
466535
a.isCompleted
467536

468-
override def foldLeft[A, U](initial: U, after: Option[A], stream: swaydb.Stream[A, Future], drop: Int, take: Option[Int])(operation: (U, A) => U): Future[U] = {
469-
def fold(previous: A, drop: Int, currentSize: Int, previousResult: U): Future[U] =
470-
if (take.contains(currentSize))
471-
Future.successful(previousResult)
472-
else
473-
stream
474-
.next(previous)
475-
.flatMap {
476-
case Some(next) =>
477-
if (drop >= 1) {
478-
fold(next, drop - 1, currentSize, previousResult)
479-
} else {
480-
try {
481-
val newResult = operation(previousResult, next)
482-
fold(next, drop, currentSize + 1, newResult)
483-
} catch {
484-
case throwable: Throwable =>
485-
Future.failed(throwable)
486-
}
487-
}
488-
489-
case None =>
490-
Future.successful(previousResult)
491-
}
492-
493-
if (take.contains(0))
494-
Future.successful(initial)
495-
else
496-
after.map(stream.next).getOrElse(stream.headOption) flatMap {
497-
case Some(first) =>
498-
if (drop >= 1) {
499-
fold(first, drop - 1, 0, initial)
500-
} else {
501-
try {
502-
val nextResult = operation(initial, first)
503-
fold(first, drop, 1, nextResult)
504-
} catch {
505-
case throwable: Throwable =>
506-
Future.failed(throwable)
507-
}
508-
}
509-
510-
case None =>
511-
Future.successful(initial)
512-
}
513-
}
537+
override def foldLeft[A, U](initial: U, after: Option[A], stream: swaydb.Stream[A, Future], drop: Int, take: Option[Int])(operation: (U, A) => U): Future[U] =
538+
swaydb.Tag.Async.foldLeft(
539+
initial = initial,
540+
after = after,
541+
stream = stream,
542+
drop = drop,
543+
take = take,
544+
operation = operation
545+
)
514546

515547
override def collectFirst[A](previous: A, stream: swaydb.Stream[A, Future])(condition: A => Boolean): Future[Option[A]] =
516-
stream.next(previous) flatMap {
517-
case some @ Some(nextA) =>
518-
if (condition(nextA))
519-
Future.successful(some)
520-
else
521-
collectFirst(nextA, stream)(condition)
522-
523-
case None =>
524-
Future.successful(None)
525-
}
548+
swaydb.Tag.Async.collectFirst(
549+
previous = previous,
550+
stream = stream,
551+
condition = condition
552+
)
526553

527554
override def fromIO[E: IO.ExceptionHandler, A](a: IO[E, A]): Future[A] = a.toFuture
528555
}

‎monix/src/main/scala/swaydb/monix/Tag.scala‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
/*
2+
* Copyright (c) 2019 Simer Plaha (@simerplaha)
3+
*
4+
* This file is a part of SwayDB.
5+
*
6+
* SwayDB is free software: you can redistribute it and/or modify
7+
* it under the terms of the GNU Affero General Public License as
8+
* published by the Free Software Foundation, either version 3 of the
9+
* License, or (at your option) any later version.
10+
*
11+
* SwayDB is distributed in the hope that it will be useful,
12+
* but WITHOUT ANY WARRANTY; without even the implied warranty of
13+
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14+
* GNU Affero General Public License for more details.
15+
*
16+
* You should have received a copy of the GNU Affero General Public License
17+
* along with SwayDB. If not, see <https://www.gnu.org/licenses/>.
18+
*/
19+
20+
package swaydb.monix
21+
22+
import monix.eval._
23+
import swaydb.Tag.Async
24+
import swaydb.data.config.ActorConfig.QueueOrder
25+
import swaydb.{Actor, IO, Serial}
26+
27+
import scala.concurrent.Promise
28+
import scala.util.{Failure, Try}
29+
30+
object Tag {
31+
32+
implicit object MonixTaskMonad extends swaydb.Monad[Task] {
33+
override def map[A, B](a: A, f: A => B): Task[B] =
34+
Task(f(a))
35+
36+
override def flatMap[A, B](a: Task[A], f: A => Task[B]): Task[B] =
37+
a.flatMap(f)
38+
39+
override def success[A](a: A): Task[A] =
40+
Task.now(a)
41+
42+
override def failed[A](a: Throwable): Task[A] =
43+
Task.fromTry(scala.util.Failure(a))
44+
}
45+
46+
implicit def apply(implicit scheduler: monix.execution.Scheduler): swaydb.Tag.Async[Task] =
47+
new Async[Task] {
48+
49+
override def createSerial(): Serial[Task] =
50+
new Serial[Task] {
51+
52+
/**
53+
* If there another preferred way to execute tasks serially in Monix instead of using an Actor
54+
* that should be used here.
55+
*/
56+
val actor = Actor[() => Any] {
57+
(run, _) =>
58+
run()
59+
}(scheduler, QueueOrder.FIFO)
60+
61+
override def execute[F](f: => F): Task[F] = {
62+
val promise = Promise[F]
63+
actor.send(() => promise.tryComplete(Try(f)))
64+
Task.fromFuture(promise.future)
65+
}
66+
}
67+
68+
override val unit: Task[Unit] =
69+
Task.unit
70+
71+
override def none[A]: Task[Option[A]] =
72+
Task.now(Option.empty)
73+
74+
override def apply[A](a: => A): Task[A] =
75+
Task(a)
76+
77+
override def map[A, B](a: A)(f: A => B): Task[B] =
78+
Task(f(a))
79+
80+
override def flatMap[A, B](fa: Task[A])(f: A => Task[B]): Task[B] =
81+
fa.flatMap(f)
82+
83+
override def success[A](value: A): Task[A] =
84+
Task.now(value)
85+
86+
override def failure[A](exception: Throwable): Task[A] =
87+
Task.fromTry(Failure(exception))
88+
89+
override def foreach[A, B](a: A)(f: A => B): Unit =
90+
f(a)
91+
92+
def fromPromise[A](a: Promise[A]): Task[A] =
93+
Task.fromFuture(a.future)
94+
95+
override def complete[A](promise: Promise[A], a: Task[A]): Unit =
96+
promise tryCompleteWith a.runToFuture
97+
98+
def isComplete[A](a: Task[A]): Boolean =
99+
a.runToFuture.isCompleted
100+
101+
override def foldLeft[A, U](initial: U, after: Option[A], stream: swaydb.Stream[A, Task], drop: Int, take: Option[Int])(operation: (U, A) => U): Task[U] =
102+
swaydb.Tag.Async.foldLeft(
103+
initial = initial,
104+
after = after,
105+
stream = stream,
106+
drop = drop,
107+
take = take,
108+
operation = operation
109+
)
110+
111+
override def collectFirst[A](previous: A, stream: swaydb.Stream[A, Task])(condition: A => Boolean): Task[Option[A]] =
112+
swaydb.Tag.Async.collectFirst(
113+
previous = previous,
114+
stream = stream,
115+
condition = condition
116+
)
117+
118+
override def fromIO[E: IO.ExceptionHandler, A](a: IO[E, A]): Task[A] =
119+
Task.fromTry(a.toTry)
120+
}
121+
}

0 commit comments

Comments
 (0)