ch08 progress

- rename book's `simple_pipe` to `syncPipe` to be more consistent
- reorganize `syncPipe` files

src/main/scala/ch07/common/libuv/libuv.scala

@@ -15,8 +15,8 @@ object LibUV:
   type ShutdownReq = Ptr[Ptr[Byte]]
   type Connection = Ptr[Byte]
   type ConnectionCB = CFuncPtr2[TCPHandle, Int, Unit]
-  type AllocCB = CFuncPtr3[TCPHandle, CSize, Ptr[Buffer], Unit]
-  type ReadCB = CFuncPtr3[TCPHandle, CSSize, Ptr[Buffer], Unit]
+  type AllocCB = CFuncPtr3[TCPHandle, CSize, Ptr[Buffer], Unit] // CSize = Usize unsigned
+  type ReadCB = CFuncPtr3[TCPHandle, CSSize, Ptr[Buffer], Unit] // CSSize = Size signed
   type WriteCB = CFuncPtr2[WriteReq, Int, Unit]
   type ShutdownCB = CFuncPtr2[ShutdownReq, Int, Unit]
   type CloseCB = CFuncPtr1[TCPHandle, Unit]

src/main/scala/ch08/common/Pipe.scala

@@ -10,14 +10,15 @@ import concurrent.{Future, ExecutionContext, Promise}
 
 trait Pipe[T, U]:
   def feed(input: T): Unit // unimplemented
+
   val handlers = mutable.Set[Pipe[U, ?]]() // the rest is implemented
   def done(): Unit = for h <- handlers do h.done()
 
   def addDestination[V](dest: Pipe[U, V]): Pipe[U, V] =
     handlers += dest
     dest
 
-  def map[V](g: U => V): Pipe[U, V] = addDestination(SyncPipe(g))
+  def map[V](g: U => V): Pipe[U, V] = addDestination(syncPipe.SyncPipe(g))
   def mapConcat[V](g: U => Seq[V]): Pipe[U, V] = addDestination(ConcatPipe(g))
   def mapOption[V](g: U => Option[V]): Pipe[U, V] = addDestination(OptionPipe(g))
 

src/main/scala/ch08/common/SyncPipe.scala → src/main/scala/ch08/syncPipe/SyncPipe.scala

@@ -1,4 +1,5 @@
 package ch08
+package syncPipe
 
 import scalanative.unsigned.{UnsignedRichLong, UnsignedRichInt}
 import scalanative.unsafe.*
@@ -41,25 +42,29 @@ object SyncPipe:
   val allocCB: AllocCB = CFuncPtr3.fromScalaFunction:
     (client: PipeHandle, size: CSize, buffer: Ptr[Buffer]) =>
       val buf = stdlib.malloc(4096) // malloc can take Int now! No need for .toUSize
-      buffer._1 = buf
+      buffer._1 = buf // Buffer = CStruct2[Ptr[Byte], CSize]
       buffer._2 = 4096.toUSize // 0.5
 
   val readCB: ReadCB = CFuncPtr3.fromScalaFunction:
-    (handle: TCPHandle, size: CSSize, buffer: Ptr[Buffer]) =>
+    (handle: TCPHandle, size: CSSize, buffer: Ptr[Buffer]) => // book has CSize
       val pipeData = handle.asInstanceOf[Ptr[Int]]
       val pipeId = !pipeData
       println(s"read $size bytes from pipe $pipeId")
 
-      if size < 0 then
+      if size < 0 then // need CSSize = Size = signed, to compare < 0
         println("size < 0, closing")
         activeStreams -= pipeId
-        val pipeDestination = handlers(pipeId)
-        pipeDestination.done()
+        val pipeDestination = handlers(pipeId) // not in the book
+        pipeDestination.done() // not in the book
         handlers.remove(pipeId)
       else
-        val dataBuffer = stdlib.malloc(size.toUSize) // removed +1 // 0.5
-        string.strncpy(dataBuffer, buffer._1, size.toUSize) // removed +1 // 0.5
+        val stringSize = size.toUSize + 1.toUSize
+
+        val dataBuffer = stdlib.malloc(stringSize)
+        string.strncpy(dataBuffer, buffer._1, stringSize)
+
         val dataString = fromCString(dataBuffer)
         stdlib.free(dataBuffer)
+
         val pipeDestination = handlers(pipeId)
         pipeDestination.feed(dataString.trim())

src/main/scala/ch08/syncPipe/main.scala

@@ -0,0 +1,24 @@
+package ch08
+package syncPipe
+
+import scala.util.{Try, Success, Failure}
+import ch07.LibUV.*, ch07.LibUVConstants.*
+
+@main
+def run: Unit =
+  println("hello!")
+  val p = SyncPipe(0)
+  val q = p
+    .map: d =>
+      println(s"consumed $d")
+      d
+    .map: d =>
+      val parsed = Try(d.toInt)
+      println(s"parsed: $parsed")
+      parsed
+    .map:
+      case Success(i) => println(s"saw number $i")
+      case Failure(f) => println(s"error: $f")
+
+  uv_run(ch07.EventLoop.loop, UV_RUN_DEFAULT)
+  println("done")