Merge pull request #16 from SystemFw/series/0.2

Series/0.2

Author: SystemFw

.scalafmt.conf

@@ -1,4 +1 @@
-project.includeFilters = [
-  "./src/*"
-  "build.sbt"
-  ]
+align=none

README.md

@@ -14,138 +14,176 @@ To get **upperbound**, add the following line to your `build.sbt`
 libraryDependencies += "org.systemfw" %% "upperbound" % "version"
 ```
 
-## Purity
+You can find the latest version in the [releases](https://github.com/SystemFw/upperbound/releases) tab.__
+**upperbound** depends on `fs2`, `cats`, `cats-effect` and `cats-collections`.
 
+**Note:**
+
+For the time being binary compatibility is **not**
+guaranteed. This is not a problem for usage in applications (which is
+where you would mostly use a rate limiter anyway), but risky if used
+in libraries. Binary compatibility will be guaranteed in the future.
+
+## Design principles
+
+**upperbound** is an interval based limiter, which means jobs are
+started at a _constant rate_. This strategy prevents spikes in
+throughput, and makes it a very good fit for client side limiting,
+e.g. calling a rate limited API.  
 **upperbound** is completely pure, which allows for ease of reasoning
 and composability. On a practical level, this means that some
 familiarity with cats, cats-effect and fs2 is required.
 
 ## Usage
 
-**upperbound**'s main datatypes are two: `Worker` and `Limiter`.
-`Worker` is the central entity in the library, it's defined as:
+### Limiter
+
+The main entity of the library is a `Limiter`, which is defined as:
 
 ``` scala
-trait Worker[F[_]] {
+trait Limiter[F[_]] {
   def submit[A](job: F[A], priority: Int = 0): F[Unit]
-  def await[A](job: F[A], priority: Int = 0): F[A]
+
+  def interval: SignallingRef[F, FiniteDuration]
+
+  def initial: FiniteDuration
+
+  def pending: F[Int]
 }
 ```
+
 The `submit` method takes an `F[A]`, which can represent any
 program, and returns an `F[Unit]` that represents the action of
 submitting it to the limiter, with the given priority. The semantics
 of `submit` are fire-and-forget: the returned `F[Unit]` immediately
 returns, without waiting for the input `F[A]` to complete its
-execution.
-
-The `await` method takes an `F[A]`, which can represent any program,
-and returns another `F[A]` that represents the action of submitting
-the input `F` to the limiter with the given priority, and waiting
-for its result.  The semantics of `await` are blocking: the returned
-`F[A]` only completes when the input `F` has finished its
-execution. However, the blocking is only semantic, no actual threads
-are blocked by the implementation.
-
-Both `Worker.submit` and `Worker.await` are designed to be called
-concurrently: every concurrent call submits a job to `Limiter`, and
-they are then started (in order of priority) at a rate which is
-no higher then the maximum rate you specify on construction.
-A higher number indicates higher priority, and FIFO order is used in
-case there are multiple jobs with the same priority being throttled.
-
-Naturally, you need to ensure that all the places in your code that
-need rate limiting share the same instance of `Worker`. It might be
-not obvious at first how to do this purely functionally, but it's in
-fact very easy: just pass a `Worker` as a parameter. For example, if
-you have a class with two methods that need rate limiting, have the
-constructor of your class accept a `Worker`.
-
-Following this approach, _your whole program_ will end up having type
-`Worker => F[Whatever]`, and all you need now is creating a
-`Worker`. This is what `Limiter` is for:
+execution.  
+`Limiter.submit` is designed to be called concurrently: every
+concurrent call submits a job to `Limiter`, and they are then started
+(in order of priority) at a rate which is no higher then the maximum
+rate you specify on construction.  A higher number indicates higher
+priority, and FIFO order is used in case there are multiple jobs with
+the same priority being throttled.  
+`interval` is an `fs2.concurrent.SignallingRef` that allows you to
+sample, change or react to changes to the current interval between two
+tasks. Finally, `initial` and `pending` return the initial interval
+specified on creation, and the number of jobs that are queued up
+waiting to start, respectively.
+
+The `Limiter` algebra is the basic building block of the library,
+additional functionality is expressed as combinators over it.
+
+### Program construction
+
+To create a `Limiter`, use the `start` method:
 
 ``` scala
-import cats.effect.Effect
-
-trait Limiter[F[_]] {
-  def worker: Worker[F]
-  def shutDown: F[Unit])
-}
+case class Rate(n: Int, t: FiniteDuration)
 
 object Limiter {
-  def start[F[_]: Effect](maxRate: Rate)(implicit ec: ExecutionContext): F[Limiter[F]]
-  def stream[F[_]: Effect](maxRate: Rate)(implicit ec: ExecutionContext): Stream[F, Limiter[F]]
+  def start[F[_]: Concurrent: Timer](maxRate: Rate, n: Int = Int.MaxValue): Resource[F, Limiter[F]]
 }
 ```
-You should only need `Limiter` at the end of your program, to assemble
-all the parts together. Imagine your program is defined as:
 
-``` scala
-import upperbound._
-import cats.effect.IO
-
-case class YourWholeProgram(w: Worker[IO]) {
-  def doStuff: IO[Unit] = {
-     def yourLogic: IO[Whatever] = ???
-     w.submit(yourLogic)
-  }
-}
-```
-you can then do:
+`start` creates a new `Limiter` and starts processing the jobs
+submitted so it, which are started at a rate no higher than `maxRate`.
+`import upperbound.syntax.rate._` exposes the `every` syntax for creating `Rate`s:
 
 ``` scala
-import upperbound._, syntax.rate._
-import fs2.Stream
-import cats.effect.{IO, IOApp, ExitCode}
-import scala.concurrent.ExecutionContext.Implicits.global
+import upperbound.syntax.rate._
 import scala.concurrent.duration._
 
-object Main extends IOApp {
-  override def run(args: List[String]): IO[ExitCode] =
-    for {
-      limiter <- Limiter.stream[IO](100 every 1.minute)
-      res <- Stream.eval(YourWholeProgram(limiter.worker).doStuff).as(ExitCode.Success)
-    } yield res
-}
+Limiter.start[F](100 every 1.minute)
 ```
 
-Note: the `every` syntax for declaring `Rate`s requires
+Additionally, `n` enforces a bound on the maximum number of jobs
+allowed to queue up while waiting for execution. Once this number is
+reached, calling `submit` will fail the resulting `F[Unit]` with a
+`LimitReachedException`, so that you can in turn signal for
+backpressure downstream. Processing restarts as soon as the number of
+jobs waiting goes below `n` again.
+
+The reason `start` returns a `cats.effect.Resource` is so that
+processing can be stopped gracefully when the `Limiter`'s lifetime is
+over.
+To assemble your program, all the places that need limiting at the
+same rate should take a `Limiter` as an argument, which is then
+created at the end of a region of sharing (typically `main`) and
+injected via `Limiter.start(...).use` or
+`Stream.resource(Limiter.start(...)).flatMap`. If this sentence didn't
+make sense to you, it's recommended to watch [this talk](https://github.com/SystemFw/scala-italy-201).
+
+
+**Note:**
+
+It's up to you whether you want to pass the `Limiter` algebra
+implicitly (as an `F[_]: Limiter` bound) or explicitly.  
+My position is that it's ok to pass algebras implicitly _as long as
+the instance is made implicit at call site_, as close as possible to
+where it's actually injected. This avoids any problems related to
+mixing things up, and is essentially equivalent to having an instance
+of your algebra for a newtype over Kleisli.
+
+Reasonable people might disagree however, and I myself pass algebras
+around both ways, in different codebases.  
+**upperbound** is slightly skewed towards the `F[_]: Limiter` style:
+internal combinators are expressed that way, and `Limiter` has a
+summoner method to allow `Limiter[F].submit`
+
+### Await
+
+As mentioned above, `submit` has fire-and-forget semantics.
+When this is not sufficient, you can use `await`:
 
 ``` scala
-import upperbound.syntax.rate._
-import scala.concurrent.duration._
+object Limiter {
+ def await[F[_]: Concurrent: Limiter, A](job: F[A], priority: Int = 0): F[A]
+}
 ```
 
-## Testing
-One further advantage of the architecture outlined above is testability.
-In particular, you normally don't care about rate limiting in unit
-tests, but the logic you are testing might require a `Worker` when
-it's actually running. In this case, it's enough to pass in a stub
-implementation of `Worker` that contains whatever logic is needed for
-your tests. In particular, you can use `upperbound.testWorker` to get an
-instance that does no rate limiting.
+`await` looks very similar to `submit`, except its semantics are
+blocking: the returned `F[A]` only completes when `job` has
+finished its execution. Note however, that the blocking is only semantic,
+no actual threads are blocked by the implementation.
 
-## Backpressure
+### Backpressure
 
-`upperbound` gives you a mechanism for applying backpressure to the
-`Limiter` based on the result of a specific job submitted by the
-corresponding `Worker` (e.g. a REST call that got rejected upstream).
-In particular, both `submit` and `await` take an extra (optional)
-argument:
+`Limiter[F].interval` offers flexible control over the rate, which can
+be used as a mechanism for applying backpressure based on the result
+of a specific job (e.g. a REST call that got rejected upstream).  
+Although this can be implemented entirely in user land, **upperbound**
+provides some backpressure helpers and combinators out of the box.
 
 ``` scala
-def submit[A](job: F[A], priority: Int, ack: BackPressure.Ack[A])
-def await[A](job: F[A], priority: Int, ack: BackPressure.Ack[A])
+class BackPressure[F[_]: Limiter, A](job: F[A]) {
+  def withBackoff(
+    backOff: FiniteDuration => FiniteDuration,
+    ack: BackPressure.Ack[A]
+  ): F[A]
+}
+object BackPressure {
+  case class Ack[-A](slowDown: Either[Throwable, A] => Boolean)
+}
 ```
 
-`BackPressure.Ack[A]` is an alias for `Either[Throwable, A] => BackPressure`,
-where `case class BackPressure(slowDown: Boolean)` is used to assert
+`withBackoff` enriches an `F[A]` with a `Limiter` constraint with the ability to apply backpressure to the `Limiter`:  
+Every time a job signals backpressure is needed through `ack`, the `Limiter` will
+adjust its current rate by applying `backOff` to it. This means the
+rate will be adjusted by calling `backOff` repeatedly whenever
+multiple consecutive jobs signal for backpressure, and reset to its
+original value when a job signals backpressure is no longer needed.
+
+Note that since jobs submitted to the Limiter are processed
+asynchronously, rate changes will not propagate instantly when the
+rate is smaller than the job completion time. However, the rate will
+eventually converge to its most up-to-date value.
+
+`BackPressure.Ack[A]` is a wrapper over  `Either[Throwable, A] => Boolean`,
+and it's used to assert
 that backpressure is needed based on a specific result (or error) of
-the submitted job. You can write your own `Ack`s, but `upperbound` provides
-some for you:
+the submitted job. You can write your own `Ack`s, but the library provides
+some for you, including:
 
-- `BackPressure.never`: never signal backpressure. If you don't
-  specify an `ack`, this is passed as a default.
 - `BackPressure.onAllErrors`: signal backpressure every time a job
   fails with any error.
 - `BackPressure.onError[E <: Throwable]`: signal backpressure if a job
@@ -154,33 +192,33 @@ some for you:
 - `BackPressure.onResult(cond: A => Boolean)`: signal backpressure
   when the result of a job satisfies the given condition.
 
-To deal with backpressure, `Limiter.start` takes extra optional parameters:
-
+Note that `withBackoff` only transforms the input job, you still need
+to actually `submit` or `await` yourself. This is done to allow
+further combinators to operate on a job as a chain of `F[A] => F[A]`
+functions before actually submitting to the `Limiter`.  
+It's also available as syntax:
+  
 ``` scala
-def start(maxRate: Rate,
-          backOff: FiniteDuration => FiniteDuration = identity,
-          n: Int = Int.MaxValue): F[Limiter]
-```
+import scala.concurrent.duration._
+import upperbound._
+import upperbound.syntax.backpressure._
 
-Every time a job signals backpressure is needed, the Limiter will
-adjust its current rate by applying `backOff` to it. This means the
-rate will be adjusted by calling `backOff` repeatedly whenever
-multiple consecutive jobs signal for backpressure, and reset to its
-original value when a job signals backpressure is no longer needed.
 
-Note that since jobs submitted to the Limiter are processed
-asynchronously, rate changes will not propagate instantly when the
-rate is smaller than the job completion time. However, the rate will
-eventually converge to its most up-to-date value.
+def prog[F[_]: Limiter, A](fa: F[A]): F[Unit] = 
+  Limiter[F].submit {
+    fa.withBackoff(_ + 1.second, Backpressure.onAllErrors)
+  }
+```
 
-Similarly, `n` allows you to place a bound on the maximum number of
-jobs allowed to queue up while waiting for execution. Once this number
-is reached, the `F` returned by any call to the corresponding
-`Worker` will immediately fail with a `LimitReachedException`, so
-that you can in turn signal for backpressure downstream. Processing
-restarts as soon as the number of jobs waiting goes below `n` again.
 
-Please be aware that the backpressure support doesn't interfere with
+Finally, please be aware that the backpressure support doesn't interfere with
 your own error handling, nor does any error handling (e.g. retrying)
-for you. This is an application specific concern and should be handled
-in application code.
+for you.
+
+
+### Test limiter
+
+If you need to satisfy a `Limiter` constraint in test code, where you
+don't actally care about rate limiting, you can use `Limiter.noOp`,
+which gives you a stub `Limiter` with no actual rate limiting and a
+synchronous `submit` method.

build.sbt

@@ -12,8 +12,9 @@ lazy val root = (project in file(".")).settings(
 lazy val commonSettings = Seq(
   organization := "org.systemfw",
   name := "upperbound",
-  scalaVersion := "2.11.11",
-  crossScalaVersions := Seq("2.11.11", "2.12.1")
+  scalaVersion := "2.12.6",
+  crossScalaVersions := Seq("2.11.12", scalaVersion.value),
+  scalafmtOnCompile := true
 )
 
 lazy val consoleSettings = Seq(
@@ -38,7 +39,7 @@ lazy val compilerOptions =
   )
 
 lazy val typeSystemEnhancements =
-  addCompilerPlugin("org.spire-math" %% "kind-projector" % "0.9.3")
+  addCompilerPlugin("org.spire-math" %% "kind-projector" % "0.9.8")
 
 def dep(org: String)(version: String)(modules: String*) =
   Seq(modules: _*) map { name =>
@@ -47,32 +48,22 @@ def dep(org: String)(version: String)(modules: String*) =
 
 lazy val dependencies =
   libraryDependencies ++= Seq(
-    "co.fs2" %% "fs2-core" % "1.0.0",
+    "co.fs2" %% "fs2-core" % "1.0.2",
     "org.typelevel" %% "cats-core" % "1.4.0",
-    "org.typelevel" %% "dogs-core" % "0.6.10",
-    "org.typelevel" %% "cats-effect" % "1.0.0"
+    "org.typelevel" %% "cats-effect" % "1.1.0",
+    "org.typelevel" %% "cats-collections-core" % "0.7.0"
   )
 
 lazy val tests = {
-  val dependencies = {
-    val specs2 = dep("org.specs2")("3.8.9")(
-      "specs2-core",
-      "specs2-scalacheck"
-    )
-
-    val mixed = Seq(
-      "org.scalacheck" %% "scalacheck" % "1.13.4",
-      "org.scalactic" %% "scalactic" % "3.0.1"
-    )
-
+  val dependencies =
     libraryDependencies ++= Seq(
-      specs2,
-      mixed
-    ).flatten.map(_ % "test")
-  }
+      "org.scalacheck" %% "scalacheck" % "1.13.4",
+      "org.scalatest" %% "scalatest" % "3.0.5",
+      "org.typelevel" %% "cats-effect-laws" % "1.0.0"
+    ).map(_ % "test")
 
   val frameworks =
-    testFrameworks := Seq(TestFrameworks.Specs2)
+    testFrameworks := Seq(TestFrameworks.ScalaTest)
 
   Seq(dependencies, frameworks)
 }

project/build.properties

@@ -1 +1 @@
-sbt.version=0.13.17
+sbt.version=1.1.6