Add ZStream.fromTcpSocketServer

[iravid]

A stream that binds a listening TCP socket and emits a stream of connections wrapped in ZManaged. Something like:

def fromSocketServer(<params>): ZStream[R, Throwable, ZManaged[R, Throwable, Connection]]

This ticket also involves defining the Connection representation. At a minimum, it should sport the following methods:

def read: ZStream[Any, Throwable, Byte]
def write: ZSink[Any, Throwable, Byte, Unit]

[regis-leray]

@iravid can you please explain what is the logic to emit socket connections

[iravid]

Every time a connection is made to the listening port, it is accepted and represented using ZManaged[Connection], and emitted to the stream.

[regis-leray]

oki thanks i will try to implement

[regis-leray]

Something like this @iravid ?

I was wondering if i put this function into trait ZStreamPlatformSpecificConstructors to use ZStream.fromInputStreamEffect() how can i access to ZSink.fromOutputStream() ?

 import blocking._
  import java.net.ServerSocket

  def fromTcpSocketServer[R](port: Int, host: Option[String] = None): ZStream[Blocking, Throwable, Managed[Throwable, Connection]] = {
    val server = Task{
      val serverSocket = new ServerSocket()

      serverSocket.bind(
        host.fold(new InetSocketAddress(port))(new InetSocketAddress(_, port))
      )
      serverSocket
    }

    for{
      srv <- ZStream.fromEffect(server)
      socket <- ZStream.unfoldM(srv)(s => effectBlockingIO(s.accept()).map(conn => Option(conn -> s)))
      conn = new Connection(socket)
    } yield ZManaged.make(ZIO.succeed(conn))(c => ZIO(c.close()).ignore)
  }

  private[zio] class Connection(socket: Socket) {
    def read: ZStream[Blocking, IOException, Byte] = fromInputStreamEffect(ZIO(socket.getInputStream).refineToOrDie[IOException])
    def write: ZSink[Any, Throwable, Byte, Unit] =  ???
    private[zio] def close(): Unit = socket.close()
  }

[iravid]

@regis-leray Yes, that's a good start, but we have to use NIO :-)