Merge branch 'devel' into araq

1 files changed, 23 insertions, 13 deletions

diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim
index 8c1cf6b18..281d5b848 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
-
+import os, tables, strutils, times, heapqueue, options, asyncstreams
+import asyncfutures except callSoon
 import nativesockets, net, deques
 
 export Port, SocketFlag
+export asyncfutures, asyncstreams
 
 #{.injectStmt: newGcInvariant().}
 
@@ -159,8 +160,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 +189,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 initCallSoonProc =
+  if asyncfutures.getCallSoonProc().isNil:
+    asyncfutures.setCallSoonProc(callSoon)
+
 when defined(windows) or defined(nimdoc):
   import winlean, sets, hashes
   type
@@ -237,15 +242,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
+    initCallSoonProc()
+
+  proc getGlobalDispatcher*(): PDispatcher =
+    if gDisp.isNil:
+      setGlobalDispatcher(newDispatcher())
+    result = gDisp
 
   proc register*(fd: AsyncFD) =
     ## Registers ``fd`` with the dispatcher.
@@ -932,14 +939,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
+    initCallSoonProc()
+
+  proc getGlobalDispatcher*(): PDispatcher =
+    if gDisp.isNil:
+      setGlobalDispatcher(newDispatcher())
+    result = gDisp
 
   proc update(fd: AsyncFD, events: set[Event]) =
     let p = getGlobalDispatcher()
@@ -1307,7 +1317,7 @@ proc recvLine*(socket: AsyncFD): Future[string] {.async, deprecated.} =
   ##
   ## **Deprecated since version 0.15.0**: Use ``asyncnet.recvLine()`` instead.
 
-  template addNLIfEmpty(): stmt =
+  template addNLIfEmpty(): untyped =
     if result.len == 0:
       result.add("\c\L")
 
@@ -1327,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)