Various changes while working on async reliability

src/core/funnel.scala

@@ -35,3 +35,9 @@ class Funnel[ItemType]():
 
 class Gun() extends Funnel[Unit]():
   def fire(): Unit = put(())
+
+def funnel[ItemType](using DummyImplicit)[ResultType](lambda: Funnel[ItemType] => ResultType)
+        : ResultType =
+
+  val funnel: Funnel[ItemType] = Funnel()
+  try lambda(funnel) finally funnel.stop()

src/core/stdio.scala

@@ -80,6 +80,7 @@ object Out:
 
 object In:
   def read(bytes: Array[Byte])(using stdio: Stdio): Int = stdio.read(bytes)
+  def close()(using stdio: Stdio): Unit = stdio.in.close()
 
 object Stdio:
   def apply
@@ -122,7 +123,7 @@ trait Stdio extends Io:
   val err: ji.PrintStream
   val in: ji.InputStream
 
-  protected[turbulence] lazy val reader: ji.Reader = ji.InputStreamReader(in, "UTF-8")
+  protected[turbulence] lazy val reader: ji.Reader = ji.InputStreamReader(in)
 
   def write(bytes: Bytes): Unit = out.write(bytes.mutable(using Unsafe), 0, bytes.length)
   def print(text: Text): Unit = out.print(text.s)

src/core/streaming.scala

@@ -173,7 +173,7 @@ object Readable:
 
   given reliableInputStream: Readable[ji.InputStream, Bytes] = in =>
     val channel: jn.channels.ReadableByteChannel = jn.channels.Channels.newChannel(in).nn
-    val buf: jn.ByteBuffer = jn.ByteBuffer.wrap(new Array[Byte](4096)).nn
+    val buf: jn.ByteBuffer = jn.ByteBuffer.wrap(new Array[Byte](1024)).nn
 
     def recur(): LazyList[Bytes] =
       try channel.read(buf) match
@@ -182,7 +182,7 @@ object Readable:
 
         case count =>
           buf.flip()
-          val size: Int = count.min(4096)
+          val size: Int = count.min(1024)
           val array: Array[Byte] = new Array[Byte](size)
           buf.get(array)
           buf.clear()
@@ -195,7 +195,7 @@ object Readable:
 
   given inputStream(using streamCut: Raises[StreamError]): Readable[ji.InputStream, Bytes] = in =>
     val channel: jn.channels.ReadableByteChannel = jn.channels.Channels.newChannel(in).nn
-    val buf: jn.ByteBuffer = jn.ByteBuffer.wrap(new Array[Byte](4096)).nn
+    val buf: jn.ByteBuffer = jn.ByteBuffer.wrap(new Array[Byte](1024)).nn
 
     def recur(total: Long): LazyList[Bytes] =
       try channel.read(buf) match
@@ -204,7 +204,7 @@ object Readable:
 
         case count =>
           buf.flip()
-          val size: Int = count.min(4096)
+          val size: Int = count.min(1024)
           val array: Array[Byte] = new Array[Byte](size)
           buf.get(array)
           buf.clear()