Merge pull request #34 from Clonkk/fix_async_bug

Fix async (see #33)

Author: Araq

examples/ex10_async_pushpull.nim

@@ -6,7 +6,7 @@ import ../zmq
 const N_TASK = 5
 
 proc pusher(nTask: int): Future[void] {.async.} =
-  var pusher = listen("tcp://*:5571", PUSH)
+  var pusher = listen("tcp://127.0.0.1:6309", PUSH)
   defer: pusher.close()
 
   for i in 1..nTask:
@@ -17,7 +17,7 @@ proc pusher(nTask: int): Future[void] {.async.} =
     await pusher.sendAsync(task)
 
 proc puller(id: int): Future[void] {.async.} =
-  const connStr = "tcp://localhost:5571"
+  const connStr = "tcp://localhost:6309"
 
   echo fmt"puller {id}: connecting to {connStr}"
   var puller = connect(connStr, PULL)
@@ -31,11 +31,10 @@ proc puller(id: int): Future[void] {.async.} =
 
 when isMainModule:
   echo "ex10_async_pushpull.nim"
-  asyncCheck pusher(N_TASK)
-
-  for i in 1..1:
-    asyncCheck puller(i)
-
+  var p = pusher(N_TASK)
+  asyncCheck puller(1)
+  waitFor p
+  # Gives time to the asyncdispatch loop to execute
   while hasPendingOperations():
-    poll()
+    drain()
 

tests/config.nims

@@ -1 +1,2 @@
 switch("path", "$projectDir/..")
+switch("threads", "on")

tests/tzmq.nim

@@ -234,10 +234,45 @@ proc asyncpoll() =
     check(rec_count == (i div 2))
     check(rec_count == snd_count)
 
+proc async_pub_sub() =
+  const N_MSGS = 10
+
+  proc publisher {.async.} =
+    var publisher = zmq.listen("tcp://127.0.0.1:5571", PUB)
+    defer: publisher.close()
+    sleep(150) # Account for slow joiner pattern
+
+    var n = 0
+    while n < N_MSGS:
+      publisher.send("topic", SNDMORE)
+      publisher.send("test " & $n)
+      await sleepAsync(100)
+      inc n
+
+  proc subscriber : Future[int] {.async.} =
+    var subscriber = zmq.connect("tcp://127.0.0.1:5571", SUB)
+    defer: subscriber.close()
+    sleep(150) # Account for slow joiner pattern
+    subscriber.setsockopt(SUBSCRIBE, "")
+    var count = 0
+    while count < N_MSGS:
+      var msg = await subscriber.receiveAsync()
+      # echo msg
+      inc(count)
+    result = count
+
+  let p = publisher()
+  let s = subscriber()
+  waitFor p
+  let count = waitFor s
+  test "async pub_sub":
+    check count == N_mSGS
+
 when isMainModule:
   reqrep()
   pubsub()
   inproc_sharectx()
   routerdealer()
   pairpair()
+  async_pub_sub()
   asyncpoll()

zmq/asynczmq.nim

@@ -70,22 +70,24 @@ proc receiveAsync*(conn: ZConnection): Future[string] =
   ## See https://github.com/zeromq/libzmq/issues/2941 and https://github.com/zeromq/pyzmq/issues/1411
   let fut = newFuture[string]("receiveAsync")
   result = fut
+  let sock = conn.socket
 
   proc cb(fd: AsyncFD): bool {.closure, gcsafe.} =
+    # the cb should work on the low level socket and not the ZConnection object
     result = true
 
     # ignore if already finished
     if fut.finished: return
 
     try:
-      let status = getsockopt[cint](conn, ZSockOptions.EVENTS)
+      let status = getsockopt[cint](sock, ZSockOptions.EVENTS)
       if (status and ZMQ_POLLIN) == 0:
         # waiting for messages
         addRead(fd, cb)
       else:
         # ready to read
         unregister(fd)
-        fut.complete conn.receive(DONTWAIT)
+        fut.complete sock.receive(DONTWAIT)
     except:
       unregister(fd)
       fut.fail getCurrentException()
@@ -103,6 +105,7 @@ proc sendAsync*(conn: ZConnection, msg: string, flags: ZSendRecvOptions = DONTWA
   ## See https://github.com/zeromq/libzmq/issues/2941 and https://github.com/zeromq/pyzmq/issues/1411
   let fut = newFuture[void]("sendAsync")
   result = fut
+  let sock = conn.socket
 
   let status = getsockopt[cint](conn, ZSockOptions.EVENTS)
   if (status and ZMQ_POLLOUT) == 0:
@@ -114,12 +117,12 @@ proc sendAsync*(conn: ZConnection, msg: string, flags: ZSendRecvOptions = DONTWA
       if fut.finished: return
 
       try:
-        let status = getsockopt[cint](conn, ZSockOptions.EVENTS)
+        let status = getsockopt[cint](sock, ZSockOptions.EVENTS)
         if (status and ZMQ_POLLOUT) == 0:
           # waiting for messages
           addWrite(fd, cb)
         else:
-          conn.send(msg, flags)
+          sock.send(msg, flags)
           unregister(fd)
           fut.complete()
       except:

zmq/connections.nim

@@ -1,5 +1,5 @@
 import ./bindings
-import std/strformat
+import std/[strformat]
 
 # Unofficial easier-for-Nim API
 
@@ -129,7 +129,7 @@ proc getsockopt*[T: SomeOrdinal|string](c: ZConnection, option: ZSockOptions): T
   Destructor
 ]#
 when defined(gcDestructors):
-  proc close*(c: var ZConnection, linger: int = 0)
+  proc close*(c: var ZConnection, linger: int = 500)
   proc `=destroy`(x: var ZConnection) =
     if x.alive:
       raise newException(ZmqError, &"Connection from/to {x.sockaddr} was destroyed but not closed.")
@@ -235,7 +235,7 @@ proc listen*(address: string, mode: ZSocketType): ZConnection =
   result = listen(address, mode, ctx)
   result.ownctx = true
 
-proc close*(c: var ZConnection, linger: int = 0) =
+proc close*(c: var ZConnection, linger: int = 500) =
   ## Closes the ``ZConnection``.
   ## Set socket linger to ``linger`` to drop buffered message and avoid blocking, then close the socket.
   ##