os/pipe.fz

# This file is part of the Fuzion language implementation.
#
# The Fuzion language implementation is free software: you can redistribute it
# and/or modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, version 3 of the License.
#
# The Fuzion language implementation is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public
# License for more details.
#
# You should have received a copy of the GNU General Public License along with The
# Fuzion language implementation.  If not, see <https://www.gnu.org/licenses/>.


# -----------------------------------------------------------------------
#
#  Tokiwa Software GmbH, Germany
#
#  Source code of Fuzion standard library feature os.pipe
#
# -----------------------------------------------------------------------


# effect for creating pipes
#
private:public pipe : effect is

  # the underlying pipe's file descriptors
  #
  pipefd array i64 := [0, 0]


  # create the pipe
  #
  check debug: pipefd.count = 2
  status := fzE_pipe_create pipefd.internal_array.data


  # has this pipe been used already?
  #
  public used concur.atomic bool := concur.atomic bool .new false





  # read bytes from a read end
  #
  read_handler (desc i64) : io.Read_Handler is
    public redef read (max_count i32) choice (Sequence u8) io.end_of_file error =>
      pipe_buffer_size := io.buffer_size.as_i32
      arr := array u8 _ pipe_buffer_size _->0
      res := fzE_pipe_read desc arr.internal_array.data arr.count
      if res = -1
        _ := fzE_pipe_close desc
        error "failed reading from $(desc)" fzE_last_error
      else if res = 0
        _ := fzE_pipe_close desc
        io.end_of_file
      else
        arr.slice 0 res


  # with writer for one pipe
  #
  with_writer(T type, LM type : mutate, w () -> outcome unit, r i64 -> T) outcome T ! concur.atomic_access
  pre
    debug: !used.read
  =>
    if status != 0
      error "creating pipe" status.as_i64
    else
      used.write true
      read_desc, write_desc := (pipefd[0], pipefd[1])

      # write bytes to a write end
      #
      writer : (io.buffered LM).Writer is
        public redef writer (bytes Sequence u8) outcome unit =>
          if fzE_pipe_write write_desc bytes.as_array.internal_array.data bytes.count = -1
            error "failed writing to $(write_desc)" fzE_last_error
          else
            unit

      match (io.buffered LM).Writer.instate writer w
        unit =>
          _ := fzE_pipe_close write_desc
          r read_desc
        e error => e


  # instate a pipe for writing and run code
  #
  public type.for_writing(T type, LM type : mutate, w () -> outcome unit, r i64 -> T) outcome T =>
    os.pipe ! ()->
      os.pipe.env.with_writer T LM w r


  # with reader for one pipe
  #
  with_reader(T type, LM type : mutate, r () -> outcome unit, w i64 -> T) outcome T ! concur.atomic_access
  pre
    debug: !used.read
  =>
    if status != 0
      error "creating pipe" status.as_i64
    else
      used.write true
      read_desc, write_desc := (pipefd[0], pipefd[1])

      match (io.buffered LM).reader (read_handler read_desc) ! r
        unit =>
          _ := fzE_pipe_close read_desc
          w write_desc
        e error => e


  # instate a pipe for reading and run code
  #
  public type.for_reading(T type, LM type : mutate, r () -> outcome unit, w i64 -> T) outcome T =>
    os.pipe ! ()->
      os.pipe.env.with_reader T LM r w


  # close all remaining pipes
  #
  public redef finally unit =>
    for fd in pipefd
    do
      if fd != 0
        _ := fzE_pipe_close fd
      else
        unit # ignore