Merge pull request #32 from Clonkk/feat_async_poller

Feat async poller

Author: Araq

tests/tzmq.nim

@@ -1,5 +1,6 @@
 import ../zmq
 import std/[unittest, os]
+import std/[asyncdispatch, asyncfutures]
 
 proc reqrep() =
   test "reqrep":
@@ -52,7 +53,7 @@ proc pubsub() =
     block alltopic:
       let topic = broadcast.receive()
       let msg = broadcast.receive()
-      check  topic == topic1
+      check topic == topic1
       check msg == "content1"
     block s1:
       let topic = sub1.receive()
@@ -66,7 +67,7 @@ proc pubsub() =
     block alltopic:
       let topic = broadcast.receive()
       let msg = broadcast.receive()
-      check  topic == topic2
+      check topic == topic2
       check msg == "content2"
     block s2:
       let topic = sub2.receive()
@@ -80,7 +81,7 @@ proc pubsub() =
     block alltopic:
       let topic = broadcast.receive()
       let msg = broadcast.receive()
-      check  topic == ""
+      check topic == ""
       check msg == "content3"
 
 proc routerdealer() =
@@ -181,10 +182,62 @@ proc pairpair() =
     for p in pairs.mitems:
       p.close()
 
+proc asyncDummy(i: int) {.async.} =
+  # echo "asyncDummy=", i
+  asyncCheck sleepAsync(2500)
+
+proc asyncpoll() =
+  test "asyncZPoller":
+    const zaddr = "tcp://127.0.0.1:15571"
+    var pusher = listen(zaddr, PUSH)
+    var puller = connect(zaddr, PULL)
+    var poller: AsyncZPoller
+    # Register the callback
+    poller.register(
+      puller,
+      ZMQ_POLLIN,
+      proc(x: ZSocket) =
+        let msg = x.receive()
+        # debugecho "==> Received msg=", msg
+        sleep(300) # Do Stuff
+    )
+
+    let N = 10
+    var snd_count = 0
+    # A client send some message
+    for i in 0..<N:
+      if (i mod 2) == 0:
+        # Can periodically send stuff
+        pusher.send($i)
+        inc(snd_count)
+    var
+      i = 0
+      rec_count = 0
+
+    while i < N:
+      # I don't recommend a high timeout because it's going to poll for the duration if there is no message in queue
+      var fut = poller.pollAsync(1)
+      # Can do Asyncstuff here
+      asyncCheck asyncDummy(i)
+      fut.addCallback proc(x: Future[int]) =
+        if x.read() > 0:
+          inc(rec_count)
+      inc(i)
+
+    # No longer polling but some callback may not have finished
+    while hasPendingOperations():
+      drain()
+
+    pusher.close()
+    puller.close()
+    check(i == N)
+    check(rec_count == (i div 2))
+    check(rec_count == snd_count)
+
 when isMainModule:
   reqrep()
   pubsub()
   inproc_sharectx()
   routerdealer()
   pairpair()
-
+  asyncpoll()

zmq.nimble

@@ -1,6 +1,6 @@
 # Package
 
-version       = "1.2.1"
+version       = "1.3.0"
 author        = "Andreas Rumpf"
 description   = "ZeroMQ wrapper"
 license       = "MIT"
@@ -14,7 +14,7 @@ task buildexamples, "Compile all examples":
       echo fstr
       if fstr.endsWith(".nim") and fstr.startsWith("./ex"):
         echo "running ", fstr
-        selfExec("cpp -d:release " & fstr)
+        selfExec("cpp --mm:orc -d:release " & fstr)
 
 task gendoc, "Generate documentation":
   exec("nimble doc --project zmq.nim --out:docs/")

zmq/asynczmq.nim

@@ -1,6 +1,65 @@
-import std/asyncdispatch
+import std/[asyncdispatch]
 import ./connections
 import ./bindings
+import ./poller
+
+type
+  AsyncZPollCB* = proc(x: ZSocket) {.gcsafe.}
+  AsyncZPoller* = object
+    ## Experimental type to use zmq.poll() with Nim's async dispatch loop
+    cb* : seq[AsyncZPollCB]
+    zpoll*: ZPoller
+
+iterator items*(poller: AsyncZPoller): tuple[item: ZPollItem, cb: AsyncZPollCB] =
+  var
+    i = 0
+    n = poller.zpoll.len()
+  while i < n:
+    yield(poller.zpoll[i], poller.cb[i])
+    inc(i)
+
+proc `=destroy`*(obj: var AsyncZPoller) =
+  if hasPendingOperations():
+    raise newException(ZmqError, "AsyncZPoller closed with pending operation")
+
+proc register*(poller: var AsyncZPoller, sock: ZSocket, event: int, cb: AsyncZPollCB) =
+  ## Register ZSocket function
+  poller.zpoll.register(sock, event)
+  poller.cb.add(cb)
+
+proc register*(poller: var AsyncZPoller, conn: ZConnection, event: int, cb: AsyncZPollCB) =
+  ## Register ZConnection
+  poller.register(conn.socket, event, cb)
+
+proc register*(poller: var AsyncZPoller, item: ZPollItem, cb: AsyncZPollCB) =
+  ## Register ZConnection
+  poller.zpoll.items.add(item)
+  poller.cb.add(cb)
+
+proc initZPoller*(poller: sink ZPoller, cb: AsyncZPollCB) : AsyncZPoller =
+  for p in poller.items:
+    result.register(p, cb)
+
+proc initZPoller*(args: openArray[tuple[item: ZConnection, cb: AsyncZPollCB]], event: cshort): AsyncZPoller =
+  ## Init a ZPoller with all items on the same event
+  for arg in args:
+    result.register(arg.item, event, arg.cb)
+
+proc pollAsync*(poller: AsyncZPoller, timeout: int = 1) : Future[int] =
+  ## Experimental API. Poll all the ZConnection and execute an async CB when ``event`` occurs.
+  result = newFuture[int]("pollAsync")
+  var r = poller.zpoll.poll(timeout)
+  # ZMQ can't have a timeout smaller than one
+  if r > 0:
+    for zpoll, cb in poller.items():
+      if events(zpoll):
+        proc localcb = cb(zpoll.socket)
+        callSoon localcb
+  if hasPendingOperations():
+    # poll vs drain ?
+    poll(timeout)
+
+  result.complete(r)
 
 proc receiveAsync*(conn: ZConnection): Future[string] =
   ## Similar to `receive()`, but `receiveAsync()` allows other async tasks to run.

zmq/bindings.nim

@@ -172,8 +172,8 @@ proc sanity_check_libzmq(): void =
   version(actual_lib_major, actual_lib_minor, actual_lib_patch)
 
   let
-    expected_lib_version = make_dotted_version(ZMQ_VERSION_MAJOR,
-        ZMQ_VERSION_MINOR, ZMQ_VERSION_PATCH)
+    # expected_lib_version = make_dotted_version(ZMQ_VERSION_MAJOR,
+        # ZMQ_VERSION_MINOR, ZMQ_VERSION_PATCH)
     actual_lib_version = make_dotted_version(actual_lib_major, actual_lib_minor, actual_lib_patch)
 
   # This is possibly over-particular about versioning
@@ -406,7 +406,7 @@ const
   ZMQ_MORE* = 1
   ZMQ_SRCFD* = 2
   ZMQ_SHARED* = 3
-type ZMsgOptions = enum
+type ZMsgOptions* = enum
   MORE = 1
   SRCFD = 2
   SHARED = 3

zmq/connections.nim

@@ -132,14 +132,14 @@ when defined(gcDestructors):
   proc close*(c: var ZConnection, linger: int = 0)
   proc `=destroy`(x: var ZConnection) =
     if x.alive:
-      raise newException(ZmqError, "Connection destroyed but not closed")
+      raise newException(ZmqError, &"Connection from/to {x.sockaddr} was destroyed but not closed.")
 
 #[
   Connect / Listen / Close
 ]#
 proc reconnect*(conn: ZConnection) =
   ## Reconnect a previously binded/connected address
-  if connect(conn.socket, conn.sockaddr) != 0:
+  if connect(conn.socket, conn.sockaddr.cstring) != 0:
     zmqError()
 
 proc reconnect*(conn: var ZConnection, address: string) =
@@ -150,12 +150,12 @@ proc reconnect*(conn: var ZConnection, address: string) =
 
 proc disconnect*(conn: ZConnection) =
   ## Disconnect the socket
-  if disconnect(conn.socket, conn.sockaddr) != 0:
+  if disconnect(conn.socket, conn.sockaddr.cstring) != 0:
     zmqError()
 
 proc unbind*(conn: ZConnection) =
   ## Unbind the socket
-  if unbind(conn.socket, conn.sockaddr) != 0:
+  if unbind(conn.socket, conn.sockaddr.cstring) != 0:
     zmqError()
 
 proc bindAddr*(conn: var ZConnection, address: string) =