From cd197e8be3ab85cc29bda63d1147dc8822520693 Mon Sep 17 00:00:00 2001
From: Calvin Rose <calsrose@gmail.com>
Date: Mon, 6 Jul 2020 19:13:32 -0500
Subject: [PATCH] Add ev/call.

This is a common operation, and making fibers manually can be tedious.
---
 examples/tcpserver.janet | 34 ++++++++++++----------------------
 src/boot/boot.janet      |  5 +++++
 src/core/ev.c            | 18 ++++++++++++++++--
 src/core/net.c           |  2 +-
 4 files changed, 34 insertions(+), 25 deletions(-)

diff --git a/examples/tcpserver.janet b/examples/tcpserver.janet
index 58e6386d..9fac15d1 100644
--- a/examples/tcpserver.janet
+++ b/examples/tcpserver.janet
@@ -1,27 +1,17 @@
 (defn handler
   "Simple handler for connections."
   [stream]
-  (def id (gensym))
-  (def b @"")
-  (print "Connection " id "!")
-  (while (:read stream 1024 b)
-    (repeat 10 (print "work for " id " ...") (ev/sleep 1))
-    (:write stream b)
-    (buffer/clear b))
-  (printf "Done %v!" id))
-
-(defn spawn-kid
-  "Run handler in a new fiber"
-  [conn]
-  (def f (fiber/new handler))
-  (ev/go f conn))
-
-(print "Starting echo server on 127.0.0.1:8000")
-
-(def server (net/server "127.0.0.1" "8000"))
+  (defer (:close stream)
+    (def id (gensym))
+    (def b @"")
+    (print "Connection " id "!")
+    (while (:read stream 1024 b)
+      (repeat 10 (print "work for " id " ...") (ev/sleep 1))
+      (:write stream b)
+      (buffer/clear b))
+    (printf "Done %v!" id)))
 
 # Run server.
-(while true
-  (with [conn (:accept server)]
-    (spawn-kid conn)
-    (ev/sleep 0.1)))
+(let [server (net/server "127.0.0.1" "8000")]
+  (print "Starting echo server on 127.0.0.1:8000")
+  (while true (ev/call handler (:accept server))))
diff --git a/src/boot/boot.janet b/src/boot/boot.janet
index d20cce3a..3f1a94a4 100644
--- a/src/boot/boot.janet
+++ b/src/boot/boot.janet
@@ -516,6 +516,11 @@
   (with-syms [iter]
     ~(do (var ,iter ,n) (while (> ,iter 0) ,;body (-- ,iter)))))
 
+(defmacro forever
+  "Evaluate body repeatedly forever, or until a break statement at the top level."
+  [& body]
+  ~(while true ,;body))
+
 (defmacro each
   "Loop over each value in ds. Returns nil."
   [x ds & body]
diff --git a/src/core/ev.c b/src/core/ev.c
index 262b942d..f07125ee 100644
--- a/src/core/ev.c
+++ b/src/core/ev.c
@@ -517,7 +517,7 @@ void janet_ev_deinit(void) {
 
 /* C functions */
 
-static Janet cfun_ev_spawn(int32_t argc, Janet *argv) {
+static Janet cfun_ev_go(int32_t argc, Janet *argv) {
     janet_arity(argc, 1, 2);
     JanetFiber *fiber = janet_getfiber(argv, 0);
     Janet value = argc == 2 ? argv[1] : janet_wrap_nil();
@@ -525,6 +525,14 @@ static Janet cfun_ev_spawn(int32_t argc, Janet *argv) {
     return argv[0];
 }
 
+static Janet cfun_ev_call(int32_t argc, Janet *argv) {
+    janet_arity(argc, 1, -1);
+    JanetFunction *fn = janet_getfunction(argv, 0);
+    JanetFiber *fiber = janet_fiber(fn, 64, argc - 1, argv + 1);
+    janet_schedule(fiber, janet_wrap_nil());
+    return janet_wrap_fiber(fiber);
+}
+
 static Janet cfun_ev_sleep(int32_t argc, Janet *argv) {
     janet_fixarity(argc, 1);
     double sec = janet_getnumber(argv, 0);
@@ -537,7 +545,13 @@ static Janet cfun_ev_sleep(int32_t argc, Janet *argv) {
 
 static const JanetReg ev_cfuns[] = {
     {
-        "ev/go", cfun_ev_spawn,
+        "ev/call", cfun_ev_call,
+        JDOC("(ev/call fn & args)\n\n"
+             "Call a function asynchronously. Returns a fiber that is scheduled to "
+             "run the function.")
+    },
+    {
+        "ev/go", cfun_ev_go,
         JDOC("(ev/go fiber &opt value)\n\n"
              "Put a fiber on the event loop to be resumed later. Optionally pass "
              "a value to resume with, otherwise resumes with nil.")
diff --git a/src/core/net.c b/src/core/net.c
index 6370dea1..aaca9d67 100644
--- a/src/core/net.c
+++ b/src/core/net.c
@@ -664,7 +664,7 @@ static void check_stream_flag(JanetStream *stream, int flag) {
         if (flag == JANET_STREAM_WRITABLE) msg = "writable";
         if (flag == JANET_STREAM_ACCEPTABLE) msg = "server";
         if (flag == JANET_STREAM_UDPSERVER) msg = "datagram server";
-        janet_panicf("bad stream - expected %s stream", msg);
+        janet_panicf("bad stream, expected %s stream", msg);
     }
 }