diff --git a/CHANGELOG.md b/CHANGELOG.md
index d55b9d41..8572e550 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,7 @@
 All notable changes to this project will be documented in this file.
 
 ## Unreleased - ???
+- Add `ev/to-stream`
 - Make `ffi/write` append to a buffer instead of insert at 0 by default.
 - Add `os/getpid` to get the current process id.
 - Add `:out` option to `os/spawn` to be able to redirect stderr to stdout with pipes.
diff --git a/src/core/ev.c b/src/core/ev.c
index bb71d7a4..3fc3a495 100644
--- a/src/core/ev.c
+++ b/src/core/ev.c
@@ -353,21 +353,22 @@ JanetStream *janet_stream(JanetHandle handle, uint32_t flags, const JanetMethod
 
 static void janet_stream_close_impl(JanetStream *stream) {
     stream->flags |= JANET_STREAM_CLOSED;
+    int canclose = !(stream->flags & JANET_STREAM_NOT_CLOSEABLE);
 #ifdef JANET_WINDOWS
     if (stream->handle != INVALID_HANDLE_VALUE) {
 #ifdef JANET_NET
         if (stream->flags & JANET_STREAM_SOCKET) {
-            closesocket((SOCKET) stream->handle);
+            if (canclose) closesocket((SOCKET) stream->handle);
         } else
 #endif
         {
-            CloseHandle(stream->handle);
+            if (canclose) CloseHandle(stream->handle);
         }
         stream->handle = INVALID_HANDLE_VALUE;
     }
 #else
     if (stream->handle != -1) {
-        close(stream->handle);
+        if (canclose) close(stream->handle);
         stream->handle = -1;
 #ifdef JANET_EV_POLL
         uint32_t i = stream->index;
@@ -3486,6 +3487,36 @@ JANET_CORE_FN(janet_cfun_ev_all_tasks,
     return janet_wrap_array(array);
 }
 
+JANET_CORE_FN(janet_cfun_to_stream,
+              "(ev/to-stream file)",
+              "Convert a core/file to a core/stream. On POSIX operating systems, this will mark "
+              "the underlying open file description as non-blocking.") {
+    janet_fixarity(argc, 1);
+    int32_t flags = 0;
+    int32_t stream_flags = 0;
+    FILE *file = janet_getfile(argv, 0, &flags);
+    if (flags & JANET_FILE_READ) stream_flags |= JANET_STREAM_READABLE;
+    if (flags & JANET_FILE_WRITE) stream_flags |= JANET_STREAM_WRITABLE;
+    if (flags & JANET_FILE_NOT_CLOSEABLE) stream_flags |= JANET_STREAM_NOT_CLOSEABLE;
+    if (flags & JANET_FILE_CLOSED) janet_panic("file is closed");
+#ifdef JANET_WINDOWS
+    int fno = _fileno(file);
+    int dupped_fno = _dup(fno);
+    if (dupped_fno == -1) janet_panic(janet_strerror(errno));
+    JanetStream *stream = janet_stream(_get_osfhandle(dupped_fno), stream_flags, NULL);
+#else
+    int handle = fileno(file);
+    int dupped_handle = 0;
+    int status = 0;
+    RETRY_EINTR(dupped_handle, dup(handle));
+    if (status == -1) janet_panic(janet_strerror(errno));
+    RETRY_EINTR(status, fcntl(dupped_handle, F_SETFL, O_NONBLOCK));
+    if (status == -1) janet_panic(janet_strerror(errno));
+    JanetStream *stream = janet_stream(dupped_handle, stream_flags, NULL);
+#endif
+    return janet_wrap_abstract(stream);
+}
+
 void janet_lib_ev(JanetTable *env) {
     JanetRegExt ev_cfuns_ext[] = {
         JANET_CORE_REG("ev/give", cfun_channel_push),
@@ -3517,6 +3548,7 @@ void janet_lib_ev(JanetTable *env) {
         JANET_CORE_REG("ev/release-rlock", janet_cfun_rwlock_read_release),
         JANET_CORE_REG("ev/release-wlock", janet_cfun_rwlock_write_release),
         JANET_CORE_REG("ev/to-file", janet_cfun_to_file),
+        JANET_CORE_REG("ev/to-stream", janet_cfun_to_stream),
         JANET_CORE_REG("ev/all-tasks", janet_cfun_ev_all_tasks),
         JANET_REG_END
     };
diff --git a/src/include/janet.h b/src/include/janet.h
index 1954af7b..672552d1 100644
--- a/src/include/janet.h
+++ b/src/include/janet.h
@@ -596,6 +596,7 @@ typedef void *JanetAbstract;
 #define JANET_STREAM_WRITABLE 0x400
 #define JANET_STREAM_ACCEPTABLE 0x800
 #define JANET_STREAM_UDPSERVER 0x1000
+#define JANET_STREAM_NOT_CLOSEABLE 0x2000
 #define JANET_STREAM_TOCLOSE 0x10000
 
 typedef enum {