syncchannel.scala

来自「JAVA 语言的函数式编程扩展」· SCALA 代码 · 共 78 行

/*                     __                                               *\
**     ________ ___   / /  ___     Scala API                            **
**    / __/ __// _ | / /  / _ |    (c) 2003-2006, LAMP/EPFL             **
**  __\ \/ /__/ __ |/ /__/ __ |                                         **
** /____/\___/_/ |_/____/_/ | |                                         **
**                          |/                                          **
\*                                                                      */

// $Id: SyncChannel.scala 14693 2008-04-17 16:04:55Z phaller $


package scala.concurrent

/** A <code>SyncChannel</code> allows one to exchange data
 *  synchronously between a reader and a writer thread.
 *  The writer thread is blocked until the data to be written
 *  has been read by a corresponding reader thread.
 *
 *  @author  Philipp Haller
 *  @version 2.0, 04/17/2008
 */
class SyncChannel[A] {

  private var pendingWrites = List[(A, SyncVar[Boolean])]()
  private var pendingReads  = List[SyncVar[A]]()

  def write(data: A) {
    // create write request
    val writeReq = new SyncVar[Boolean]

    this.synchronized {
      // check whether there is a reader waiting
      if (!pendingReads.isEmpty) {
        val readReq  = pendingReads.head
        pendingReads = pendingReads.tail

        // let reader continue
        readReq set data

        // resolve write request
        writeReq set true
      }
      else {
        // enqueue write request
        pendingWrites = pendingWrites ::: List((data, writeReq))
      }
    }

    writeReq.get
  }

  def read: A = {
    // create read request
    val readReq = new SyncVar[A]

    this.synchronized {
      // check whether there is a writer waiting
      if (!pendingWrites.isEmpty) {
        // read data
        val (data, writeReq) = pendingWrites.head
        pendingWrites = pendingWrites.tail

        // let writer continue
        writeReq set true

        // resolve read request
        readReq set data
      }
      else {
        // enqueue read request
        pendingReads = pendingReads ::: List(readReq)
      }
    }

    readReq.get
  }
}