added JANET_STREAM_CLOSING to indicate the stream should be closed at the end of the event loop handling

Author: zevv

src/core/ev.c

@@ -1502,6 +1502,9 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp to) {
                     state = state->_next;
                 }
             }
+            if(stream->flags |= JANET_STREAM_CLOSING) {
+                janet_stream_close(stream);
+            }
         }
     }
 }
@@ -1656,6 +1659,9 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
                     janet_unlisten(state, 0);
                 state = next_state;
             }
+            if(stream->flags |= JANET_STREAM_CLOSING) {
+                janet_stream_close(stream);
+            }
         }
     }
 }
@@ -1854,6 +1860,9 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
 
                 state = next_state;
             }
+            if(stream->flags |= JANET_STREAM_CLOSING) {
+                janet_stream_close(stream);
+            }
         }
     }
 }
@@ -1970,6 +1979,9 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
                 status3 == JANET_ASYNC_STATUS_DONE ||
                 status4 == JANET_ASYNC_STATUS_DONE)
             janet_unlisten(state, 0);
+        if(state->stream->flags |= JANET_STREAM_CLOSING) {
+            janet_stream_close(state->stream);
+        }
     }
 }
 
@@ -2495,11 +2507,11 @@ static JanetAsyncStatus handle_connect(JanetListenerState *s) {
         if (res == 0) {
             janet_schedule(s->fiber, janet_wrap_abstract(s->stream));
         } else {
-            // TODO help needed. janet_stream_close(s->stream);
+            s->stream->flags |= JANET_STREAM_CLOSING;
             janet_cancel(s->fiber, janet_cstringv(strerror(res)));
         }
     } else {
-        // TODO help needed. janet_stream_close(s->stream);
+        s->stream->flags |= JANET_STREAM_CLOSING;
         janet_cancel(s->fiber, janet_ev_lasterr());
     }
     return JANET_ASYNC_STATUS_DONE;

src/include/janet.h

@@ -564,6 +564,7 @@ typedef void *JanetAbstract;
 #define JANET_STREAM_CLOSED 0x1
 #define JANET_STREAM_SOCKET 0x2
 #define JANET_STREAM_IOCP 0x4
+#define JANET_STREAM_CLOSING 0x8
 #define JANET_STREAM_READABLE 0x200
 #define JANET_STREAM_WRITABLE 0x400
 #define JANET_STREAM_ACCEPTABLE 0x800