(* buffer.sml *)

(* CML buffers with adjustable maximum lengths *)

structure Buffer :> BUFFER =
struct

open CML

type 'a buffer =
       {addCh  : 'a chan,            (* values sent to this channel will
                                        be added to buffer *)
        delCh  : 'a chan,            (* values read from this channel will
                                        be deleted from buffer *)
        lenCh  : int chan,           (* channel on which buffer's current
                                        length can be read *)
        newMaxLenReqCh : int chan,   (* channel on which a proposed new
                                        maximum length for buffer can be
                                        sent *)
        newMaxLenRepCh : bool chan}  (* channel from which the
                                        success/failure of an attempt
                                        to change buffer's maximum
                                        length can be received *)

(* val server : 'a buffer * int -> 'b

   in a call server(buf, maxLen), we require that maxLen >= 1 *)

fun server({addCh, delCh, lenCh, newMaxLenReqCh, newMaxLenRepCh},
           maxLen) =
      let (* val serv : Fifo.fifo * int -> 'b

             in a call serv(queue, maxLen), we require that
             Fifo.length queue <= maxLen *)

          fun serv(queue, maxLen) =
                select
                [(* handle additions *)

                 if Fifo.length queue < maxLen
                 then wrap(recvEvt addCh,
                           fn x => serv(Fifo.enqueue(queue, x), maxLen))
                 else never,

                 (* handle deletions *)

                 if Fifo.length queue > 0
                 then let val (queue, x) = Fifo.dequeue queue
                      in wrap(sendEvt(delCh, x),
                              fn () => serv(queue, maxLen))
                      end
                 else never,

                 (* handle requests for current length of buffer *)

                 wrap(sendEvt(lenCh, Fifo.length queue),
                      fn () => serv(queue, maxLen)),

                 (* handle requests to change maximum length of buffer *)

                 wrap(recvEvt newMaxLenReqCh,
                      fn newMaxLen =>
                           if Fifo.length queue <= newMaxLen
                           then (send(newMaxLenRepCh, true);
                                 serv(queue, newMaxLen))
                           else (send(newMaxLenRepCh, false);
                                 serv(queue, maxLen)))]
      in serv(Fifo.empty, maxLen) end

exception Size

fun make maxLen =
      if maxLen <= 0
      then raise Size
      else let val buf =
                     {addCh          = channel(),
                      delCh          = channel(),
                      lenCh          = channel(),
                      newMaxLenReqCh = channel(),
                      newMaxLenRepCh = channel()}
           in spawn(fn () => server(buf, maxLen)); buf end

fun addEvt ({addCh, ...} : 'a buffer) x = sendEvt(addCh, x)

fun delEvt({delCh, ...} : 'a buffer) = recvEvt delCh

fun lenEvt({lenCh, ...} : 'a buffer) = recvEvt lenCh

fun newMaxLenEvt ({newMaxLenReqCh, newMaxLenRepCh, ...} : 'a buffer) n =
      wrap(sendEvt(newMaxLenReqCh, n),
           fn () => recv newMaxLenRepCh)

end;