changed net/connect to be non-blocking / asynchronous

Author: zevv

src/core/ev.c

@@ -2456,7 +2456,8 @@ void janet_ev_recvfrom(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, in
 typedef enum {
     JANET_ASYNC_WRITEMODE_WRITE,
     JANET_ASYNC_WRITEMODE_SEND,
-    JANET_ASYNC_WRITEMODE_SENDTO
+    JANET_ASYNC_WRITEMODE_SENDTO,
+    JANET_ASYNC_WRITEMODE_CONNECT
 } JanetWriteMode;
 
 typedef struct {
@@ -2509,6 +2510,25 @@ JanetAsyncStatus ev_machine_write(JanetListenerState *s, JanetAsyncEvent event)
         }
         break;
         case JANET_ASYNC_EVENT_USER: {
+#ifdef JANET_NET
+            if (state->mode == JANET_ASYNC_WRITEMODE_CONNECT) {
+                int res = 0;
+                int size = sizeof res;
+                int r = getsockopt((SOCKET)s->stream->handle, SOL_SOCKET, SO_ERROR, (char *)&res, &size);
+                if (r == 0) {
+                    if (res == 0) {
+                        janet_schedule(s->fiber, janet_wrap_abstract(s->stream));
+                    } else {
+                        // TODO JSOCKCLOSE(s->sock) but is in net.c;
+                        janet_cancel(s->fiber, janet_cstringv(strerror(res)));
+                    }
+                    return JANET_ASYNC_STATUS_DONE;
+                } else {
+                    // TODO panic? assert?
+                    return JANET_ASYNC_STATUS_NOT_DONE;
+                }
+            }
+#endif
             /* Begin write */
             int32_t len;
             const uint8_t *bytes;
@@ -2572,6 +2592,25 @@ JanetAsyncStatus ev_machine_write(JanetListenerState *s, JanetAsyncEvent event)
             janet_cancel(s->fiber, janet_cstringv("stream hup"));
             return JANET_ASYNC_STATUS_DONE;
         case JANET_ASYNC_EVENT_WRITE: {
+#ifdef JANET_NET
+            if (state->mode == JANET_ASYNC_WRITEMODE_CONNECT) {
+                int res = 0;
+                socklen_t size = sizeof res;
+                int r = getsockopt(s->stream->handle, SOL_SOCKET, SO_ERROR, &res, &size);
+                if (r == 0) {
+                    if (res == 0) {
+                        janet_schedule(s->fiber, janet_wrap_abstract(s->stream));
+                    } else {
+                        // TODO JSOCKCLOSE(s->sock) but is in net.c;
+                        janet_cancel(s->fiber, janet_cstringv(strerror(res)));
+                    }
+                    return JANET_ASYNC_STATUS_DONE;
+                } else {
+                    // TODO panic? assert?
+                    return JANET_ASYNC_STATUS_NOT_DONE;
+                }
+            }
+#endif
             int32_t start, len;
             const uint8_t *bytes;
             start = state->start;
@@ -2674,6 +2713,10 @@ void janet_ev_sendto_buffer(JanetStream *stream, JanetBuffer *buf, void *dest, i
 void janet_ev_sendto_string(JanetStream *stream, JanetString str, void *dest, int flags) {
     janet_ev_write_generic(stream, (void *) str, dest, JANET_ASYNC_WRITEMODE_SENDTO, 0, flags);
 }
+
+void janet_ev_connect(JanetStream *stream, int flags) {
+    janet_ev_write_generic(stream, NULL, NULL, JANET_ASYNC_WRITEMODE_CONNECT, 0, flags);
+}
 #endif
 
 /* For a pipe ID */

src/core/net.c

@@ -477,32 +477,43 @@ JANET_CORE_FN(cfun_net_connect,
         }
     }
 
+    /* Wrap socket in abstract type JanetStream */
+    JanetStream *stream = make_stream(sock, JANET_STREAM_READABLE | JANET_STREAM_WRITABLE);
+
+    /* Set the socket to non-blocking mode */
+    janet_net_socknoblock(sock);
+
     /* Connect to socket */
 #ifdef JANET_WINDOWS
     int status = WSAConnect(sock, addr, addrlen, NULL, NULL, NULL, NULL);
-    Janet lasterr = janet_ev_lasterr();
+    int err = WSAGetLastError();
     freeaddrinfo(ai);
 #else
     int status = connect(sock, addr, addrlen);
-    Janet lasterr = janet_ev_lasterr();
+    int err = errno;
     if (is_unix) {
         janet_free(ai);
     } else {
         freeaddrinfo(ai);
     }
 #endif
 
-    if (status == -1) {
-        JSOCKCLOSE(sock);
-        janet_panicf("could not connect socket: %V", lasterr);
+    if (status != 0) {
+#ifdef JANET_WINDOWS
+        if(err != WSAEWOULDBLOCK) {
+#else
+        if(err != EINPROGRESS) {
+#endif
+            JSOCKCLOSE(sock);
+            Janet lasterr = janet_ev_lasterr();
+            janet_panicf("could not connect socket: %V", lasterr);
+        }
     }
+   
+    /* Handle the connect() result in the event loop*/
+    janet_ev_connect(stream, MSG_NOSIGNAL);
 
-    /* Set up the socket for non-blocking IO after connect - TODO - non-blocking connect? */
-    janet_net_socknoblock(sock);
-
-    /* Wrap socket in abstract type JanetStream */
-    JanetStream *stream = make_stream(sock, JANET_STREAM_READABLE | JANET_STREAM_WRITABLE);
-    return janet_wrap_abstract(stream);
+    janet_await();
 }
 
 static const char *serverify_socket(JSock sfd) {

src/include/janet.h

@@ -1479,6 +1479,7 @@ JANET_API void janet_ev_readchunk(JanetStream *stream, JanetBuffer *buf, int32_t
 JANET_API void janet_ev_recv(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags);
 JANET_API void janet_ev_recvchunk(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags);
 JANET_API void janet_ev_recvfrom(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags);
+JANET_API void janet_ev_connect(JanetStream *stream, int flags);
 #endif
 
 /* Write async to a stream */