From e86863e8f55d862b7690fbe6a9cca87ebf425913 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Zieli=C5=84ski?= <michal@zielinscy.org.pl>
Date: Mon, 5 Jun 2017 00:12:44 +0200
Subject: [PATCH 1/5] asyncdispatch: split asyncfutures into its own module

This slightly changes behaviour of callSoon - before loop is initialized, callSoon will call the function immediately.
---
 lib/pure/asyncdispatch.nim               | 31 +++++++++++++++---------
 lib/pure/{includes => }/asyncfutures.nim | 12 ++++++++-
 lib/upcoming/asyncdispatch.nim           | 30 +++++++++++++++--------
 tests/async/tasyncrecursion.nim          |  1 +
 4 files changed, 52 insertions(+), 22 deletions(-)
 rename lib/pure/{includes => }/asyncfutures.nim (97%)

diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim
index 8c1cf6b18..59218acce 100644
--- a/lib/pure/asyncdispatch.nim
+++ b/lib/pure/asyncdispatch.nim
@@ -9,11 +9,11 @@
 
 include "system/inclrtl"
 
-import os, tables, strutils, times, heapqueue, options
-
+import os, tables, strutils, times, heapqueue, options, asyncfutures
 import nativesockets, net, deques
 
 export Port, SocketFlag
+export asyncfutures
 
 #{.injectStmt: newGcInvariant().}
 
@@ -159,8 +159,6 @@ export Port, SocketFlag
 
 # TODO: Check if yielded future is nil and throw a more meaningful exception
 
-include includes/asyncfutures
-
 type
   PDispatcherBase = ref object of RootRef
     timers*: HeapQueue[tuple[finishAt: float, fut: Future[void]]]
@@ -190,6 +188,12 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} =
       result = int((timerTimeout - curTime) * 1000)
       if result < 0: result = 0
 
+proc callSoon*(cbproc: proc ()) {.gcsafe.}
+
+proc initGlobalDispatcher =
+  if asyncfutures.callSoonProc == nil:
+    asyncfutures.callSoonProc = callSoon
+
 when defined(windows) or defined(nimdoc):
   import winlean, sets, hashes
   type
@@ -237,15 +241,17 @@ when defined(windows) or defined(nimdoc):
     result.callbacks = initDeque[proc ()](64)
 
   var gDisp{.threadvar.}: PDispatcher ## Global dispatcher
-  proc getGlobalDispatcher*(): PDispatcher =
-    ## Retrieves the global thread-local dispatcher.
-    if gDisp.isNil: gDisp = newDispatcher()
-    result = gDisp
 
   proc setGlobalDispatcher*(disp: PDispatcher) =
     if not gDisp.isNil:
       assert gDisp.callbacks.len == 0
     gDisp = disp
+    initGlobalDispatcher()
+
+  proc getGlobalDispatcher*(): PDispatcher =
+    if gDisp.isNil:
+      setGlobalDispatcher(newDispatcher())
+    result = gDisp
 
   proc register*(fd: AsyncFD) =
     ## Registers ``fd`` with the dispatcher.
@@ -932,14 +938,17 @@ else:
     result.callbacks = initDeque[proc ()](64)
 
   var gDisp{.threadvar.}: PDispatcher ## Global dispatcher
-  proc getGlobalDispatcher*(): PDispatcher =
-    if gDisp.isNil: gDisp = newDispatcher()
-    result = gDisp
 
   proc setGlobalDispatcher*(disp: PDispatcher) =
     if not gDisp.isNil:
       assert gDisp.callbacks.len == 0
     gDisp = disp
+    initGlobalDispatcher()
+
+  proc getGlobalDispatcher*(): PDispatcher =
+    if gDisp.isNil:
+      setGlobalDispatcher(newDispatcher())
+    result = gDisp
 
   proc update(fd: AsyncFD, events: set[Event]) =
     let p = getGlobalDispatcher()
diff --git a/lib/pure/includes/asyncfutures.nim b/lib/pure/asyncfutures.nim
similarity index 97%
rename from lib/pure/includes/asyncfutures.nim
rename to lib/pure/asyncfutures.nim
index 6af5bf3cf..8a1538546 100644
--- a/lib/pure/includes/asyncfutures.nim
+++ b/lib/pure/asyncfutures.nim
@@ -1,3 +1,6 @@
+include "system/inclrtl"
+
+import os, tables, strutils, times, heapqueue, options, deques
 
 # TODO: This shouldn't need to be included, but should ideally be exported.
 type
@@ -30,7 +33,14 @@ type
 when not defined(release):
   var currentID = 0
 
-proc callSoon*(cbproc: proc ()) {.gcsafe.}
+var callSoonProc* {.threadvar.}: (proc(cbproc: proc ()) {.gcsafe.})
+
+proc callSoon(cbproc: proc ()) =
+  if callSoonProc == nil:
+    # Loop not initialized yet. Call the function directly to allow setup code to use futures.
+    cbproc()
+  else:
+    callSoonProc(cbproc)
 
 template setupFutureBase(fromProc: string) =
   new(result)
diff --git a/lib/upcoming/asyncdispatch.nim b/lib/upcoming/asyncdispatch.nim
index 1623d8375..84c33f3ab 100644
--- a/lib/upcoming/asyncdispatch.nim
+++ b/lib/upcoming/asyncdispatch.nim
@@ -9,11 +9,12 @@
 
 include "system/inclrtl"
 
-import os, tables, strutils, times, heapqueue, lists, options
+import os, tables, strutils, times, heapqueue, lists, options, asyncfutures
 
 import nativesockets, net, deques
 
 export Port, SocketFlag
+export asyncfutures
 
 #{.injectStmt: newGcInvariant().}
 
@@ -130,8 +131,6 @@ export Port, SocketFlag
 
 # TODO: Check if yielded future is nil and throw a more meaningful exception
 
-include "../includes/asyncfutures"
-
 type
   PDispatcherBase = ref object of RootRef
     timers: HeapQueue[tuple[finishAt: float, fut: Future[void]]]
@@ -161,6 +160,12 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} =
       result = int((timerTimeout - curTime) * 1000)
       if result < 0: result = 0
 
+proc callSoon*(cbproc: proc ()) {.gcsafe.}
+
+proc initGlobalDispatcher =
+  if asyncfutures.callSoonProc == nil:
+    asyncfutures.callSoonProc = callSoon
+
 when defined(windows) or defined(nimdoc):
   import winlean, sets, hashes
   type
@@ -214,15 +219,17 @@ when defined(windows) or defined(nimdoc):
     result.callbacks = initDeque[proc ()](64)
 
   var gDisp{.threadvar.}: PDispatcher ## Global dispatcher
-  proc getGlobalDispatcher*(): PDispatcher =
-    ## Retrieves the global thread-local dispatcher.
-    if gDisp.isNil: gDisp = newDispatcher()
-    result = gDisp
 
   proc setGlobalDispatcher*(disp: PDispatcher) =
     if not gDisp.isNil:
       assert gDisp.callbacks.len == 0
     gDisp = disp
+    initGlobalDispatcher()
+
+  proc getGlobalDispatcher*(): PDispatcher =
+    if gDisp.isNil:
+      setGlobalDispatcher(newDispatcher())
+    result = gDisp
 
   proc register*(fd: AsyncFD) =
     ## Registers ``fd`` with the dispatcher.
@@ -1081,14 +1088,17 @@ else:
     result.callbacks = initDeque[proc ()](InitDelayedCallbackListSize)
 
   var gDisp{.threadvar.}: PDispatcher ## Global dispatcher
-  proc getGlobalDispatcher*(): PDispatcher =
-    if gDisp.isNil: gDisp = newDispatcher()
-    result = gDisp
 
   proc setGlobalDispatcher*(disp: PDispatcher) =
     if not gDisp.isNil:
       assert gDisp.callbacks.len == 0
     gDisp = disp
+    initGlobalDispatcher()
+
+  proc getGlobalDispatcher*(): PDispatcher =
+    if gDisp.isNil:
+      setGlobalDispatcher(newDispatcher())
+    result = gDisp
 
   proc register*(fd: AsyncFD) =
     let p = getGlobalDispatcher()
diff --git a/tests/async/tasyncrecursion.nim b/tests/async/tasyncrecursion.nim
index 54482edab..1aeebe9b4 100644
--- a/tests/async/tasyncrecursion.nim
+++ b/tests/async/tasyncrecursion.nim
@@ -17,5 +17,6 @@ proc asyncRecursionTest*(): Future[int] {.async.} =
     inc(i)
 
 when isMainModule:
+  setGlobalDispatcher(newDispatcher())
   var i = waitFor asyncRecursionTest()
   echo i

From 93827e6ab8effab38696f1d1c25bf65c2f538675 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Zieli=C5=84ski?= <michal@zielinscy.org.pl>
Date: Mon, 5 Jun 2017 14:44:12 +0200
Subject: [PATCH 2/5] split FutureStream from asyncfutures

---
 lib/pure/asyncdispatch.nim     |   9 +--
 lib/pure/asyncfutures.nim      | 103 +-------------------------------
 lib/pure/asyncstreams.nim      | 105 +++++++++++++++++++++++++++++++++
 lib/upcoming/asyncdispatch.nim |   9 +--
 4 files changed, 117 insertions(+), 109 deletions(-)
 create mode 100644 lib/pure/asyncstreams.nim

diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim
index 59218acce..2a412d967 100644
--- a/lib/pure/asyncdispatch.nim
+++ b/lib/pure/asyncdispatch.nim
@@ -9,11 +9,12 @@
 
 include "system/inclrtl"
 
-import os, tables, strutils, times, heapqueue, options, asyncfutures
+import os, tables, strutils, times, heapqueue, options, asyncstreams
+import asyncfutures except callSoon
 import nativesockets, net, deques
 
 export Port, SocketFlag
-export asyncfutures
+export asyncfutures, asyncstreams
 
 #{.injectStmt: newGcInvariant().}
 
@@ -188,7 +189,7 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} =
       result = int((timerTimeout - curTime) * 1000)
       if result < 0: result = 0
 
-proc callSoon*(cbproc: proc ()) {.gcsafe.}
+proc callSoon(cbproc: proc ()) {.gcsafe.}
 
 proc initGlobalDispatcher =
   if asyncfutures.callSoonProc == nil:
@@ -1336,7 +1337,7 @@ proc recvLine*(socket: AsyncFD): Future[string] {.async, deprecated.} =
       return
     add(result, c)
 
-proc callSoon*(cbproc: proc ()) =
+proc callSoon(cbproc: proc ()) =
   ## Schedule `cbproc` to be called as soon as possible.
   ## The callback is called when control returns to the event loop.
   getGlobalDispatcher().callbacks.addLast(cbproc)
diff --git a/lib/pure/asyncfutures.nim b/lib/pure/asyncfutures.nim
index 8a1538546..f7d96d5e4 100644
--- a/lib/pure/asyncfutures.nim
+++ b/lib/pure/asyncfutures.nim
@@ -19,12 +19,6 @@ type
 
   FutureVar*[T] = distinct Future[T]
 
-  FutureStream*[T] = ref object of FutureBase   ## Special future that acts as
-                                                ## a queue. Its API is still
-                                                ## experimental and so is
-                                                ## subject to change.
-    queue: Deque[T]
-
   FutureError* = object of Exception
     cause*: FutureBase
 
@@ -35,7 +29,7 @@ when not defined(release):
 
 var callSoonProc* {.threadvar.}: (proc(cbproc: proc ()) {.gcsafe.})
 
-proc callSoon(cbproc: proc ()) =
+proc callSoon*(cbproc: proc ()) =
   if callSoonProc == nil:
     # Loop not initialized yet. Call the function directly to allow setup code to use futures.
     cbproc()
@@ -66,22 +60,6 @@ proc newFutureVar*[T](fromProc = "unspecified"): FutureVar[T] =
   ## that this future belongs to, is a good habit as it helps with debugging.
   result = FutureVar[T](newFuture[T](fromProc))
 
-proc newFutureStream*[T](fromProc = "unspecified"): FutureStream[T] =
-  ## Create a new ``FutureStream``. This future's callback is activated when
-  ## two events occur:
-  ##
-  ## * New data is written into the future stream.
-  ## * The future stream is completed (this means that no more data will be
-  ##   written).
-  ##
-  ## Specifying ``fromProc``, which is a string specifying the name of the proc
-  ## that this future belongs to, is a good habit as it helps with debugging.
-  ##
-  ## **Note:** The API of FutureStream is still new and so has a higher
-  ## likelihood of changing in the future.
-  setupFutureBase(fromProc)
-  result.queue = initDeque[T]()
-
 proc clean*[T](future: FutureVar[T]) =
   ## Resets the ``finished`` status of ``future``.
   Future[T](future).finished = false
@@ -148,12 +126,6 @@ proc complete*[T](future: FutureVar[T], val: T) =
   if not fut.cb.isNil():
     fut.cb()
 
-proc complete*[T](future: FutureStream[T]) =
-  ## Completes a ``FutureStream`` signalling the end of data.
-  future.finished = true
-  if not future.cb.isNil():
-    future.cb()
-
 proc fail*[T](future: Future[T], error: ref Exception) =
   ## Completes ``future`` with ``error``.
   #assert(not future.finished, "Future already finished, cannot finish twice.")
@@ -190,20 +162,6 @@ proc `callback=`*[T](future: Future[T],
   ## If future has already completed then ``cb`` will be called immediately.
   future.callback = proc () = cb(future)
 
-proc `callback=`*[T](future: FutureStream[T],
-    cb: proc (future: FutureStream[T]) {.closure,gcsafe.}) =
-  ## Sets the callback proc to be called when data was placed inside the
-  ## future stream.
-  ##
-  ## The callback is also called when the future is completed. So you should
-  ## use ``finished`` to check whether data is available.
-  ##
-  ## If the future stream already has data or is finished then ``cb`` will be
-  ## called immediately.
-  future.cb = proc () = cb(future)
-  if future.queue.len > 0 or future.finished:
-    callSoon(future.cb)
-
 proc injectStacktrace[T](future: Future[T]) =
   # TODO: Come up with something better.
   when not defined(release):
@@ -250,18 +208,12 @@ proc mget*[T](future: FutureVar[T]): var T =
   ## Future has not been finished.
   result = Future[T](future).value
 
-proc finished*[T](future: Future[T] | FutureVar[T] | FutureStream[T]): bool =
+proc finished*[T](future: Future[T] | FutureVar[T]): bool =
   ## Determines whether ``future`` has completed.
   ##
   ## ``True`` may indicate an error or a value. Use ``failed`` to distinguish.
-  ##
-  ## For a ``FutureStream`` a ``true`` value means that no more data will be
-  ## placed inside the stream _and_ that there is no data waiting to be
-  ## retrieved.
   when future is FutureVar[T]:
     result = (Future[T](future)).finished
-  elif future is FutureStream[T]:
-    result = future.finished and future.queue.len == 0
   else:
     result = future.finished
 
@@ -269,57 +221,6 @@ proc failed*(future: FutureBase): bool =
   ## Determines whether ``future`` completed with an error.
   return future.error != nil
 
-proc write*[T](future: FutureStream[T], value: T): Future[void] =
-  ## Writes the specified value inside the specified future stream.
-  ##
-  ## This will raise ``ValueError`` if ``future`` is finished.
-  result = newFuture[void]("FutureStream.put")
-  if future.finished:
-    let msg = "FutureStream is finished and so no longer accepts new data."
-    result.fail(newException(ValueError, msg))
-    return
-  # TODO: Implement limiting of the streams storage to prevent it growing
-  # infinitely when no reads are occuring.
-  future.queue.addLast(value)
-  if not future.cb.isNil: future.cb()
-  result.complete()
-
-proc read*[T](future: FutureStream[T]): Future[(bool, T)] =
-  ## Returns a future that will complete when the ``FutureStream`` has data
-  ## placed into it. The future will be completed with the oldest
-  ## value stored inside the stream. The return value will also determine
-  ## whether data was retrieved, ``false`` means that the future stream was
-  ## completed and no data was retrieved.
-  ##
-  ## This function will remove the data that was returned from the underlying
-  ## ``FutureStream``.
-  var resFut = newFuture[(bool, T)]("FutureStream.take")
-  let savedCb = future.cb
-  future.callback =
-    proc (fs: FutureStream[T]) =
-      # We don't want this callback called again.
-      future.cb = nil
-
-      # The return value depends on whether the FutureStream has finished.
-      var res: (bool, T)
-      if finished(fs):
-        # Remember, this callback is called when the FutureStream is completed.
-        res[0] = false
-      else:
-        res[0] = true
-        res[1] = fs.queue.popFirst()
-
-      if not resFut.finished:
-        resFut.complete(res)
-
-      # If the saved callback isn't nil then let's call it.
-      if not savedCb.isNil: savedCb()
-  return resFut
-
-proc len*[T](future: FutureStream[T]): int =
-  ## Returns the amount of data pieces inside the stream.
-  future.queue.len
-
 proc asyncCheck*[T](future: Future[T]) =
   ## Sets a callback on ``future`` which raises an exception if the future
   ## finished with an error.
diff --git a/lib/pure/asyncstreams.nim b/lib/pure/asyncstreams.nim
new file mode 100644
index 000000000..d3ea143f3
--- /dev/null
+++ b/lib/pure/asyncstreams.nim
@@ -0,0 +1,105 @@
+import asyncfutures
+
+import deques
+
+type
+  FutureStream*[T] = ref object   ## Special future that acts as
+                                  ## a queue. Its API is still
+                                  ## experimental and so is
+                                  ## subject to change.
+    queue: Deque[T]
+    finished: bool
+    cb: proc () {.closure, gcsafe.}
+
+proc newFutureStream*[T](fromProc = "unspecified"): FutureStream[T] =
+  ## Create a new ``FutureStream``. This future's callback is activated when
+  ## two events occur:
+  ##
+  ## * New data is written into the future stream.
+  ## * The future stream is completed (this means that no more data will be
+  ##   written).
+  ##
+  ## Specifying ``fromProc``, which is a string specifying the name of the proc
+  ## that this future belongs to, is a good habit as it helps with debugging.
+  ##
+  ## **Note:** The API of FutureStream is still new and so has a higher
+  ## likelihood of changing in the future.
+  result = FutureStream[T](finished: false, cb: nil)
+  result.queue = initDeque[T]()
+
+proc complete*[T](future: FutureStream[T]) =
+  ## Completes a ``FutureStream`` signalling the end of data.
+  future.finished = true
+  if not future.cb.isNil:
+    future.cb()
+
+proc `callback=`*[T](future: FutureStream[T],
+    cb: proc (future: FutureStream[T]) {.closure,gcsafe.}) =
+  ## Sets the callback proc to be called when data was placed inside the
+  ## future stream.
+  ##
+  ## The callback is also called when the future is completed. So you should
+  ## use ``finished`` to check whether data is available.
+  ##
+  ## If the future stream already has data or is finished then ``cb`` will be
+  ## called immediately.
+  future.cb = proc () = cb(future)
+  if future.queue.len > 0 or future.finished:
+    callSoon(future.cb)
+
+proc finished*[T](future: FutureStream[T]): bool =
+  ## Check if a ``FutureStream`` is finished. ``true`` value means that
+  ## no more data will be placed inside the stream _and_ that there is
+  ## no data waiting to be retrieved.
+  result = future.finished and future.queue.len == 0
+
+proc write*[T](future: FutureStream[T], value: T): Future[void] =
+  ## Writes the specified value inside the specified future stream.
+  ##
+  ## This will raise ``ValueError`` if ``future`` is finished.
+  result = newFuture[void]("FutureStream.put")
+  if future.finished:
+    let msg = "FutureStream is finished and so no longer accepts new data."
+    result.fail(newException(ValueError, msg))
+    return
+  # TODO: Implement limiting of the streams storage to prevent it growing
+  # infinitely when no reads are occuring.
+  future.queue.addLast(value)
+  if not future.cb.isNil: future.cb()
+  result.complete()
+
+proc read*[T](future: FutureStream[T]): Future[(bool, T)] =
+  ## Returns a future that will complete when the ``FutureStream`` has data
+  ## placed into it. The future will be completed with the oldest
+  ## value stored inside the stream. The return value will also determine
+  ## whether data was retrieved, ``false`` means that the future stream was
+  ## completed and no data was retrieved.
+  ##
+  ## This function will remove the data that was returned from the underlying
+  ## ``FutureStream``.
+  var resFut = newFuture[(bool, T)]("FutureStream.take")
+  let savedCb = future.cb
+  future.callback =
+    proc (fs: FutureStream[T]) =
+      # We don't want this callback called again.
+      future.cb = nil
+
+      # The return value depends on whether the FutureStream has finished.
+      var res: (bool, T)
+      if finished(fs):
+        # Remember, this callback is called when the FutureStream is completed.
+        res[0] = false
+      else:
+        res[0] = true
+        res[1] = fs.queue.popFirst()
+
+      if not resFut.finished:
+        resFut.complete(res)
+
+      # If the saved callback isn't nil then let's call it.
+      if not savedCb.isNil: savedCb()
+  return resFut
+
+proc len*[T](future: FutureStream[T]): int =
+  ## Returns the amount of data pieces inside the stream.
+  future.queue.len
diff --git a/lib/upcoming/asyncdispatch.nim b/lib/upcoming/asyncdispatch.nim
index 84c33f3ab..d90ca6534 100644
--- a/lib/upcoming/asyncdispatch.nim
+++ b/lib/upcoming/asyncdispatch.nim
@@ -9,12 +9,13 @@
 
 include "system/inclrtl"
 
-import os, tables, strutils, times, heapqueue, lists, options, asyncfutures
+import os, tables, strutils, times, heapqueue, lists, options, asyncstreams
+import asyncfutures except callSoon
 
 import nativesockets, net, deques
 
 export Port, SocketFlag
-export asyncfutures
+export asyncfutures, asyncstreams
 
 #{.injectStmt: newGcInvariant().}
 
@@ -160,7 +161,7 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} =
       result = int((timerTimeout - curTime) * 1000)
       if result < 0: result = 0
 
-proc callSoon*(cbproc: proc ()) {.gcsafe.}
+proc callSoon(cbproc: proc ()) {.gcsafe.}
 
 proc initGlobalDispatcher =
   if asyncfutures.callSoonProc == nil:
@@ -1611,7 +1612,7 @@ proc recvLine*(socket: AsyncFD): Future[string] {.async.} =
       return
     add(result, c)
 
-proc callSoon*(cbproc: proc ()) =
+proc callSoon(cbproc: proc ()) =
   ## Schedule `cbproc` to be called as soon as possible.
   ## The callback is called when control returns to the event loop.
   getGlobalDispatcher().callbacks.addLast(cbproc)

From 797690ba3ff415a457499ddf0edda24c31644b1d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Zieli=C5=84ski?= <michal@zielinscy.org.pl>
Date: Mon, 12 Jun 2017 11:47:52 +0200
Subject: [PATCH 3/5] Future: support for multiple callbacks

---
 lib/pure/asyncfutures.nim  | 83 ++++++++++++++++++++++++++++----------
 tests/async/tcallbacks.nim | 20 +++++++++
 2 files changed, 82 insertions(+), 21 deletions(-)
 create mode 100644 tests/async/tcallbacks.nim

diff --git a/lib/pure/asyncfutures.nim b/lib/pure/asyncfutures.nim
index f7d96d5e4..2d86d3d8d 100644
--- a/lib/pure/asyncfutures.nim
+++ b/lib/pure/asyncfutures.nim
@@ -4,8 +4,15 @@ import os, tables, strutils, times, heapqueue, options, deques
 
 # TODO: This shouldn't need to be included, but should ideally be exported.
 type
+  CallbackFunc = proc () {.closure, gcsafe.}
+
+  CallbackList = object
+    function: CallbackFunc
+    next: ref CallbackList
+
   FutureBase* = ref object of RootObj ## Untyped future.
-    cb: proc () {.closure,gcsafe.}
+    callbacks: CallbackList
+
     finished: bool
     error*: ref Exception ## Stored exception
     errorStackTrace*: string
@@ -86,6 +93,33 @@ proc checkFinished[T](future: Future[T]) =
       err.cause = future
       raise err
 
+proc call(callbacks: var CallbackList) =
+  var current = callbacks
+
+  while true:
+    if current.function != nil:
+      callSoon(current.function)
+
+    if current.next == nil:
+      break
+    else:
+      current = current.next[]
+
+  # callback will be called only once, let GC collect them now
+  callbacks.next = nil
+  callbacks.function = nil
+
+proc add(callbacks: var CallbackList, function: CallbackFunc) =
+  if callbacks.function == nil:
+    callbacks.function = function
+    assert callbacks.next == nil
+  else:
+    let newNext = new(ref CallbackList)
+    newNext.function = callbacks.function
+    newNext.next = callbacks.next
+    callbacks.next = newNext
+    callbacks.function = function
+
 proc complete*[T](future: Future[T], val: T) =
   ## Completes ``future`` with value ``val``.
   #assert(not future.finished, "Future already finished, cannot finish twice.")
@@ -93,8 +127,7 @@ proc complete*[T](future: Future[T], val: T) =
   assert(future.error == nil)
   future.value = val
   future.finished = true
-  if future.cb != nil:
-    future.cb()
+  future.callbacks.call()
 
 proc complete*(future: Future[void]) =
   ## Completes a void ``future``.
@@ -102,8 +135,7 @@ proc complete*(future: Future[void]) =
   checkFinished(future)
   assert(future.error == nil)
   future.finished = true
-  if future.cb != nil:
-    future.cb()
+  future.callbacks.call()
 
 proc complete*[T](future: FutureVar[T]) =
   ## Completes a ``FutureVar``.
@@ -111,8 +143,7 @@ proc complete*[T](future: FutureVar[T]) =
   checkFinished(fut)
   assert(fut.error == nil)
   fut.finished = true
-  if fut.cb != nil:
-    fut.cb()
+  fut.callbacks.call
 
 proc complete*[T](future: FutureVar[T], val: T) =
   ## Completes a ``FutureVar`` with value ``val``.
@@ -134,26 +165,36 @@ proc fail*[T](future: Future[T], error: ref Exception) =
   future.error = error
   future.errorStackTrace =
     if getStackTrace(error) == "": getStackTrace() else: getStackTrace(error)
-  if future.cb != nil:
-    future.cb()
+  future.callbacks.call
+
+proc clearCallbacks(future: FutureBase) =
+  future.callbacks.function = nil
+  future.callbacks.next = nil
+
+proc addCallback*(future: FutureBase, cb: proc() {.closure,gcsafe.}) =
+  ## Adds the callbacks proc to be called when the future completes.
+  ##
+  ## If future has already completed then ``cb`` will be called immediately.
+  assert cb != nil
+  if future.finished:
+    callSoon(cb)
   else:
-    # This is to prevent exceptions from being silently ignored when a future
-    # is discarded.
-    # TODO: This may turn out to be a bad idea.
-    # Turns out this is a bad idea.
-    #raise error
-    discard
+    future.callbacks.add cb
+
+proc addCallback*[T](future: Future[T], cb: proc(future: Future[T]) {.closure,gcsafe.}) =
+  ## Adds the callbacks proc to be called when the future completes.
+  ##
+  ## If future has already completed then ``cb`` will be called immediately.
+  future.addCallback proc() = cb(future)
 
 proc `callback=`*(future: FutureBase, cb: proc () {.closure,gcsafe.}) =
-  ## Sets the callback proc to be called when the future completes.
+  ## Clears the list of callbacks and sets the callback proc to be called when the future completes.
   ##
   ## If future has already completed then ``cb`` will be called immediately.
   ##
-  ## **Note**: You most likely want the other ``callback`` setter which
-  ## passes ``future`` as a param to the callback.
-  future.cb = cb
-  if future.finished:
-    callSoon(future.cb)
+  ## It's recommended to use ``addCallback`` or ``then`` instead.
+  future.clearCallbacks
+  future.addCallback cb
 
 proc `callback=`*[T](future: Future[T],
     cb: proc (future: Future[T]) {.closure,gcsafe.}) =
diff --git a/tests/async/tcallbacks.nim b/tests/async/tcallbacks.nim
new file mode 100644
index 000000000..8c08032cd
--- /dev/null
+++ b/tests/async/tcallbacks.nim
@@ -0,0 +1,20 @@
+discard """
+  exitcode: 0
+  output: '''3
+2
+1
+5
+'''
+"""
+import asyncfutures
+
+let f1: Future[int] = newFuture[int]()
+f1.addCallback(proc() = echo 1)
+f1.addCallback(proc() = echo 2)
+f1.addCallback(proc() = echo 3)
+f1.complete(10)
+
+let f2: Future[int] = newFuture[int]()
+f2.addCallback(proc() = echo 4)
+f2.callback = proc() = echo 5
+f2.complete(10)

From 6bdd00db0ed97595fb78930976312fc987d95397 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Zieli=C5=84ski?= <michal@zielinscy.org.pl>
Date: Thu, 6 Jul 2017 13:24:33 +0200
Subject: [PATCH 4/5] asyncdispatch: add callSoon getter/setter, renames

---
 lib/pure/asyncdispatch.nim     | 10 +++++-----
 lib/pure/asyncfutures.nim      | 17 ++++++++++++++---
 lib/upcoming/asyncdispatch.nim | 10 +++++-----
 3 files changed, 24 insertions(+), 13 deletions(-)

diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim
index 2a412d967..0ce6115ad 100644
--- a/lib/pure/asyncdispatch.nim
+++ b/lib/pure/asyncdispatch.nim
@@ -191,9 +191,9 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} =
 
 proc callSoon(cbproc: proc ()) {.gcsafe.}
 
-proc initGlobalDispatcher =
-  if asyncfutures.callSoonProc == nil:
-    asyncfutures.callSoonProc = callSoon
+proc initCallSoonProc =
+  if asyncfutures.getCallSoonProc() == nil:
+    asyncfutures.setCallSoonProc(callSoon)
 
 when defined(windows) or defined(nimdoc):
   import winlean, sets, hashes
@@ -247,7 +247,7 @@ when defined(windows) or defined(nimdoc):
     if not gDisp.isNil:
       assert gDisp.callbacks.len == 0
     gDisp = disp
-    initGlobalDispatcher()
+    initCallSoonProc()
 
   proc getGlobalDispatcher*(): PDispatcher =
     if gDisp.isNil:
@@ -944,7 +944,7 @@ else:
     if not gDisp.isNil:
       assert gDisp.callbacks.len == 0
     gDisp = disp
-    initGlobalDispatcher()
+    initCallSoonProc()
 
   proc getGlobalDispatcher*(): PDispatcher =
     if gDisp.isNil:
diff --git a/lib/pure/asyncfutures.nim b/lib/pure/asyncfutures.nim
index 2d86d3d8d..8b27e7104 100644
--- a/lib/pure/asyncfutures.nim
+++ b/lib/pure/asyncfutures.nim
@@ -1,5 +1,3 @@
-include "system/inclrtl"
-
 import os, tables, strutils, times, heapqueue, options, deques
 
 # TODO: This shouldn't need to be included, but should ideally be exported.
@@ -34,9 +32,22 @@ type
 when not defined(release):
   var currentID = 0
 
-var callSoonProc* {.threadvar.}: (proc(cbproc: proc ()) {.gcsafe.})
+var callSoonProc {.threadvar.}: (proc(cbproc: proc ()) {.gcsafe.})
+
+proc getCallSoonProc*(): (proc(cbproc: proc ()) {.gcsafe.}) =
+  ## Get current implementation of ``callSoon``.
+  return callSoonProc
+
+proc setCallSoonProc*(p: (proc(cbproc: proc ()) {.gcsafe.})) =
+  ## Change current implementation of ``callSoon``. This is normally called when dispatcher from ``asyncdispatcher`` is initialized.
+  callSoonProc = p
 
 proc callSoon*(cbproc: proc ()) =
+  ## Call ``cbproc`` "soon".
+  ##
+  ## If async dispatcher is running, ``cbproc`` will be executed during next dispatcher tick.
+  ##
+  ## If async dispatcher is not running, ``cbproc`` will be executed immediately.
   if callSoonProc == nil:
     # Loop not initialized yet. Call the function directly to allow setup code to use futures.
     cbproc()
diff --git a/lib/upcoming/asyncdispatch.nim b/lib/upcoming/asyncdispatch.nim
index d90ca6534..9e9668aad 100644
--- a/lib/upcoming/asyncdispatch.nim
+++ b/lib/upcoming/asyncdispatch.nim
@@ -163,9 +163,9 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} =
 
 proc callSoon(cbproc: proc ()) {.gcsafe.}
 
-proc initGlobalDispatcher =
-  if asyncfutures.callSoonProc == nil:
-    asyncfutures.callSoonProc = callSoon
+proc initCallSoonProc =
+  if asyncfutures.getCallSoonProc() == nil:
+    asyncfutures.setCallSoonProc(callSoon)
 
 when defined(windows) or defined(nimdoc):
   import winlean, sets, hashes
@@ -225,7 +225,7 @@ when defined(windows) or defined(nimdoc):
     if not gDisp.isNil:
       assert gDisp.callbacks.len == 0
     gDisp = disp
-    initGlobalDispatcher()
+    initCallSoonProc()
 
   proc getGlobalDispatcher*(): PDispatcher =
     if gDisp.isNil:
@@ -1094,7 +1094,7 @@ else:
     if not gDisp.isNil:
       assert gDisp.callbacks.len == 0
     gDisp = disp
-    initGlobalDispatcher()
+    initCallSoonProc()
 
   proc getGlobalDispatcher*(): PDispatcher =
     if gDisp.isNil:

From cb5b1c93448a165ca2c00c461a7aa9f45990ccc2 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Zieli=C5=84ski?= <michal@zielinscy.org.pl>
Date: Wed, 12 Jul 2017 10:23:48 +0200
Subject: [PATCH 5/5] asyncdispatch: formatting fixes

---
 lib/pure/asyncdispatch.nim     |  2 +-
 lib/pure/asyncfutures.nim      | 22 +++++++++++++---------
 lib/upcoming/asyncdispatch.nim |  2 +-
 3 files changed, 15 insertions(+), 11 deletions(-)

diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim
index 0ce6115ad..28b20feaa 100644
--- a/lib/pure/asyncdispatch.nim
+++ b/lib/pure/asyncdispatch.nim
@@ -192,7 +192,7 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} =
 proc callSoon(cbproc: proc ()) {.gcsafe.}
 
 proc initCallSoonProc =
-  if asyncfutures.getCallSoonProc() == nil:
+  if asyncfutures.getCallSoonProc().isNil:
     asyncfutures.setCallSoonProc(callSoon)
 
 when defined(windows) or defined(nimdoc):
diff --git a/lib/pure/asyncfutures.nim b/lib/pure/asyncfutures.nim
index 8b27e7104..a9e97c14c 100644
--- a/lib/pure/asyncfutures.nim
+++ b/lib/pure/asyncfutures.nim
@@ -32,7 +32,7 @@ type
 when not defined(release):
   var currentID = 0
 
-var callSoonProc {.threadvar.}: (proc(cbproc: proc ()) {.gcsafe.})
+var callSoonProc {.threadvar.}: proc (cbproc: proc ()) {.gcsafe.}
 
 proc getCallSoonProc*(): (proc(cbproc: proc ()) {.gcsafe.}) =
   ## Get current implementation of ``callSoon``.
@@ -48,7 +48,7 @@ proc callSoon*(cbproc: proc ()) =
   ## If async dispatcher is running, ``cbproc`` will be executed during next dispatcher tick.
   ##
   ## If async dispatcher is not running, ``cbproc`` will be executed immediately.
-  if callSoonProc == nil:
+  if callSoonProc.isNil:
     # Loop not initialized yet. Call the function directly to allow setup code to use futures.
     cbproc()
   else:
@@ -108,10 +108,10 @@ proc call(callbacks: var CallbackList) =
   var current = callbacks
 
   while true:
-    if current.function != nil:
+    if not current.function.isNil:
       callSoon(current.function)
 
-    if current.next == nil:
+    if current.next.isNil:
       break
     else:
       current = current.next[]
@@ -121,7 +121,7 @@ proc call(callbacks: var CallbackList) =
   callbacks.function = nil
 
 proc add(callbacks: var CallbackList, function: CallbackFunc) =
-  if callbacks.function == nil:
+  if callbacks.function.isNil:
     callbacks.function = function
     assert callbacks.next == nil
   else:
@@ -154,7 +154,7 @@ proc complete*[T](future: FutureVar[T]) =
   checkFinished(fut)
   assert(fut.error == nil)
   fut.finished = true
-  fut.callbacks.call
+  fut.callbacks.call()
 
 proc complete*[T](future: FutureVar[T], val: T) =
   ## Completes a ``FutureVar`` with value ``val``.
@@ -176,7 +176,7 @@ proc fail*[T](future: Future[T], error: ref Exception) =
   future.error = error
   future.errorStackTrace =
     if getStackTrace(error) == "": getStackTrace() else: getStackTrace(error)
-  future.callbacks.call
+  future.callbacks.call()
 
 proc clearCallbacks(future: FutureBase) =
   future.callbacks.function = nil
@@ -192,11 +192,15 @@ proc addCallback*(future: FutureBase, cb: proc() {.closure,gcsafe.}) =
   else:
     future.callbacks.add cb
 
-proc addCallback*[T](future: Future[T], cb: proc(future: Future[T]) {.closure,gcsafe.}) =
+proc addCallback*[T](future: Future[T],
+                     cb: proc (future: Future[T]) {.closure,gcsafe.}) =
   ## Adds the callbacks proc to be called when the future completes.
   ##
   ## If future has already completed then ``cb`` will be called immediately.
-  future.addCallback proc() = cb(future)
+  future.addCallback(
+    proc() =
+      cb(future)
+  )
 
 proc `callback=`*(future: FutureBase, cb: proc () {.closure,gcsafe.}) =
   ## Clears the list of callbacks and sets the callback proc to be called when the future completes.
diff --git a/lib/upcoming/asyncdispatch.nim b/lib/upcoming/asyncdispatch.nim
index 9e9668aad..4e3b06173 100644
--- a/lib/upcoming/asyncdispatch.nim
+++ b/lib/upcoming/asyncdispatch.nim
@@ -164,7 +164,7 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} =
 proc callSoon(cbproc: proc ()) {.gcsafe.}
 
 proc initCallSoonProc =
-  if asyncfutures.getCallSoonProc() == nil:
+  if asyncfutures.getCallSoonProc().isNil:
     asyncfutures.setCallSoonProc(callSoon)
 
 when defined(windows) or defined(nimdoc):