Add Future implementation

Author: lbarasti

README.md

@@ -9,36 +9,90 @@ A collection of primitives for event-driven programming.
 
 ## Installation
 
-1. Add the dependency to your `shard.yml`:
+Add the dependency to your `shard.yml` and run `shards install`:
 
-   ```yaml
-   dependencies:
-     concur:
-       github: your-github-user/concur
-   ```
-
-2. Run `shards install`
+  ```yaml
+  dependencies:
+    concur:
+      github: lbarasti/concur
+  ```
 
 ## Usage
 
 ```crystal
 require "concur"
 ```
 
-TODO: Write usage instructions here
+### Using Future
+You can use `Future` to wrap asynchronous computations that might fail.
+```crystal
+f = Future.new {
+  sleep 2 # a placeholder for some expensive computation or for a lengthy IO operation
+  "Success!"
+}
+
+f.await # => "Success!"
+```
+If you want to keep on manipulating the result of a future in a separate fiber, then you can rely on `Future`'s instance methods.
+
+For example, given a future `f`, you can apply a function to the wrapped value with `#map`, filter it with `#select` and recover from failure with `#recover`
+
+```crystal
+f.map { |v| v.downcase }
+  .select { |v| v.size < 3 }
+  .recover { "default_key" }
+```
+
+Here is a contrived example to showcase some other common methods.
+
+You can combine the result of two Futures into one with `#zip`:
+
+```crystal
+author_f : Future(User) = db.user_by_id(1)
+reviewer_f : Future(User) = db.user_by_id(2)
+
+permission_f : Future(Bool) = author_f.zip(reviewer_f) { |author, reviewer|
+  author.has_reviewer?(reviewer) 
+}
+```
+
+You can use `#flat_map` to avoid nesting futures:
+
+```crystal
+content_f : Future(Content) = permission_f
+  .flat_map { |reviewer_is_allowed|
+    if reviewer_is_allowed
+      db.content_by_user(1) # => Future(Content)
+    else
+      raise NotAllowedError.new
+    end
+  }
+```
+
+And perform side effects with `#on_success` and `#on_error`.
+
+```crystal
+content_f
+  .on_success { |content|
+    reviewer_f.await!.email(content)
+  }
+  .on_error { |ex| log_error(ex) }
+```
+
+Check out the [API docs](https://lbarasti.com/concur/) for more details.
 
 ## Development
 
-TODO: Write development instructions here
+Run `shards install` to install the project dependencies. You can then run `crystal spec` to verify that all the tests are passing.
 
 ## Contributing
 
-1. Fork it (<https://github.com/your-github-user/concur/fork>)
+1. Fork it (<https://github.com/lbarasti/concur/fork>)
 2. Create your feature branch (`git checkout -b my-new-feature`)
 3. Commit your changes (`git commit -am 'Add some feature'`)
 4. Push to the branch (`git push origin my-new-feature`)
 5. Create a new Pull Request
 
 ## Contributors
 
-- [lorenzo.barasti](https://github.com/your-github-user) - creator and maintainer
+- [lbarasti](https://github.com/lbarasti) - creator and maintainer

shard.yml

@@ -1,10 +1,10 @@
 name: concur
-version: 0.1.0
+version: 0.2.0
 
 authors:
   - lbarasti <[email protected]>
 
-crystal: 1.0.0
+crystal: ">=1.0.0"
 
 dependencies:
   rate_limiter:

spec/future_spec.cr

@@ -0,0 +1,259 @@
+require "./spec_helper"
+
+describe Future do
+  it "runs a computation asyncronously" do
+    ch = Channel(Nil).new
+    
+    Future.new { ch.send nil }
+
+    ch.receive
+  end
+
+  it "can be awaited on" do
+    f = Future.new { 2** 10 }
+    res = f.await
+    res.should eq 1024
+    typeof(res).should eq (Int32 | Exception)
+  end
+
+  it "can be awaited on multiple times" do
+    f = Future.new { 2** 10 }
+    10.times { 
+      f.await.should eq 1024
+    }
+  end
+
+  it "can be awaited on by multiple fibers" do
+    n_fibers = 100
+    results = Channel(Int32 | Exception).new(n_fibers)
+    f = Future.new { sleep rand; 2** 10 }
+    n_fibers.times { 
+      spawn {
+        results.send f.await
+      }
+    }
+    n_fibers.times {
+      results.receive.should eq 1024
+    }
+  end
+
+  it "can be awaited on and raise on exception" do
+    f = Future.new { sleep 0.4; raise CustomError.new("error"); 42 }
+    expect_raises(CustomError) {
+      f.await!
+    }
+  end
+
+  it "can be awaited on with timeout" do
+    f = Future.new { sleep 0.6; 42 }
+    f.await(0.2.seconds)
+      .should be_a Concur::Timeout
+    f.await
+      .should eq 42
+    f.await(0.1.seconds)
+      .should eq 42
+  end
+
+  it "can be awaited on with timeout and raise on exception" do
+    f = Future.new { sleep 0.4; raise CustomError.new("error"); 42 }
+    expect_raises(Concur::Timeout) {
+      f.await!(0.2.seconds)
+    }
+    expect_raises(CustomError) {
+      f.await!(0.3.seconds)
+    }
+  end
+
+  it "can be queried for completion" do
+    f = Future.new { sleep 0.3; :done }
+    f.done?.should be_false
+    f.await
+    f.done?.should be_true
+  end
+
+  it "supports running callbacks, in order, on completion" do
+    results = [] of Int32
+    f = Future.new { 5 }
+    f.on_complete { |r|
+      results << r.as(Int32) + 1
+      raise Exception.new("runtime exception")
+    }.on_complete { |r|
+      case r
+      when Int32
+        results << r
+      end
+    }.await.should eq 5
+
+    results.should eq [6, 5]
+  end
+
+  it "supports running callbacks in order on success" do
+    results = [] of Int32
+    f = Future.new { 5 }
+    f.on_success { |r|
+      results << r.as(Int32) + 1
+      raise Exception.new("runtime exception")
+    }.on_success { |r|
+      case r
+      when Int32
+        results << r
+      end
+    }.await.should eq 5
+
+    results.should eq [6, 5]
+  end
+
+  it "will not run #on_success callbacks if the future fails" do
+    results = [] of Int32
+    f = Future.new { raise CustomError.new; 5 }
+    f.on_success { |r|
+      results << 1
+    }.on_success { |r|
+      raise Exception.new
+      results << 2
+    }.await.should be_a CustomError
+
+    results.should be_empty
+  end
+
+  it "supports running callbacks in order on error" do
+    results = [] of Exception
+    f = Future.new { raise CustomError.new; 5 }
+
+    f.on_error { |ex|
+      results << ex.as(CustomError)
+      raise Exception.new("runtime exception")
+    }.on_error { |ex|
+      results << Timeout.new
+    }.await.should be_a CustomError
+
+    results.first.should be_a CustomError
+    results.last.should be_a Timeout
+  end
+
+  it "will not run #on_error callbacks if the future succeeds" do
+    results = [] of Int32
+    f = Future.new { 5 }
+    f.on_error { |r|
+      results << 1
+    }.on_error { |r|
+      raise Exception.new
+      results << 2
+    }.await.should eq 5
+
+    results.should be_empty
+end
+
+  it "supports function composition" do
+    c_1 = Future.new { rand ** 2 }
+    c_2 = c_1.map { |x| 1 - x }
+    
+    (c_1.await! + c_2.await!).should eq 1
+    c_2.done?.should be_true
+  end
+
+  it "doesn't propagate errors backward when composing functions" do
+    c_1 = Future.new { 2 }
+    c_2 = c_1.map { |x| raise CustomError.new("error"); 1 - x }
+    
+    c_2.await.should be_a CustomError
+    c_1.await.should eq 2
+  end
+
+  it "propagates errors when composing functions" do
+    c_1 = Future.new { raise CustomError.new("error"); rand ** 2 }
+    c_2 = c_1.map { |x| raise Exception.new("generic error"); 1 - x }
+    
+    c_2.await.should be_a CustomError
+  end
+
+  it "can be filtered" do
+    f = Future.new { 5 }
+    g = f.select { |x| x % 2 == 1 }
+    h = f.select { |x| x % 2 == 0 }
+    g.await.should eq 5
+    h.await.should be_a Concur::EmptyError
+  end
+
+  it "supports combining two futures with #flat_map" do
+    f = Future.new { 5 }
+    g = Future.new { 3 }
+    
+    f.flat_map { |x| g.map { |y| x + y } }
+      .await.should eq 8
+  end
+
+  it "will return the first encountered error on #flat_map" do
+    f = Future.new { raise CustomError.new; 5 }
+    g = Future.new { 3 }
+    
+    f.flat_map { |x| g.map { |y| x + y } }
+      .await.should be_a CustomError
+  end
+
+  it "can be flattened in case of nested futures" do
+    f = Future.new { Future.new { 5 } }
+    f.await.should be_a Future(Int32)
+    
+    f.flatten
+      .await.should eq 5
+  end
+
+  it "supports mapping its underlying value to a specific type" do
+    f = Future.new { rand < 0.5 ? "hello" : 5 }
+    
+    typeof(f.await).should eq String | Int32 | Exception
+    v = f.map_to(Int32).await
+    typeof(v).should eq Int32 | Exception
+  end
+
+  it "raises a TypeCastError when mapping to an incompatible type" do
+    f = Future.new { rand < 0.5 ? "hello" : 5 }
+    g = f.map { |v| v.class == Int32 ? "hello 2" : 6 }
+
+    expect_raises(TypeCastError) {
+      fv = f.map_to(Int32).await!
+      gv = g.map_to(Int32).await!
+    }
+  end
+
+  it "can be recovered if it fails" do
+    f = Future.new { raise CustomError.new; 42 }
+    f.recover { |e|
+      case e
+      when CustomError
+        100
+      else
+        raise Exception.new("Unexpected failure")
+      end
+    }.await.should eq 100
+  end
+
+  it "does not recover successful futures" do
+    f = Future.new { 42 }
+    f.recover { |ex| 43 }
+      .await.should eq 42
+  end
+
+  it "can be transformed" do
+    f = Future.new { 42 }
+    g = Future.new { raise CustomError.new; 42 }
+    t = -> (res : Int32 | Exception) {
+      case res
+      when Int32
+        "#{res + 1}"
+      else
+        "0"
+      end
+    }
+    f.transform(&t).await.should eq "43"
+    g.transform(&t).await.should eq "0"
+  end
+
+  it "supports zipping futures together" do
+    f = Future.new { 1 }
+    g = Future.new { 2 }
+    f.zip(g) { |v1, v2| v1 + v2 }
+      .await.should eq 3
+  end
+end