title: How to make a FIFO Stream (Blocking Queue) thread-safe in Scala (possible deadlock)?
body: <p>Building on the answer provided at <a href="https://stackoverflow.com/questions/7553270/is-there-a-fifo-stream-in-scala">Is there a FIFO stream in Scala?</a>, I would like to know how to use the FIFOStream (renamed to QueueStream below) in practice, and concurrently. Since I'm rather new to Scala and Rx, I'm sticking with Futures so far, but I'd welcome alternatives (especially if using Futures in this way is too painful).  </p>
  
  <p>The problem is exemplified by the <code>FIXME</code> comment in the below worksheet; the comment says it doesn't run, but based on past experience I think it may be deadlocking somehow.</p>
  
  <pre><code>import java.util.concurrent.{Executors, BlockingQueue, LinkedBlockingQueue}
  import scala.collection.JavaConversions._
  import scala.concurrent.{ExecutionContext, Future}
  import scala.concurrent.duration._
  import scala.language.postfixOps
  
  implicit val ec = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(8))
  
  
  class QueueStream[A]( private val queue: BlockingQueue[Option[A]] ) {
  
    implicit protected val defaultTimeout: Duration = 100 milliseconds
  
    def toStream(q: BlockingQueue[Option[A]] = queue): Future[Stream[A]] = {
      def computeStream(q: BlockingQueue[Option[A]] = queue): Stream[A] =
        queue.take() match {
          case Some(a) =&gt; Stream cons(a, computeStream())
          case None =&gt; Stream.empty
        }
      Future {
        computeStream()
      }
    }
    def toStreamNoWait(q: BlockingQueue[Option[A]] = queue ):
    Future[Stream[A]] = {
      def computeStream(q: BlockingQueue[Option[A]] = queue): Stream[A] = {
        val timeout = implicitly[Duration]
        queue.poll(timeout.toMillis, MILLISECONDS) match {
          case Some(a) =&gt; Stream cons(a, computeStream())
          case None =&gt; Stream.empty
        }
      }
      Future {
        computeStream()
      }
    }
  
    def size () = queue.size()
    def close() = queue add None
    def enqueue( as: A* ) = queue addAll as.map( Some(_) )
  }
  object QueueStream {
    def apply[A]() = new QueueStream[A](new LinkedBlockingQueue)
  }
  def printIntsInQueue(queue: QueueStream[Int]): Unit = {
    println("About to print some Ints...")
    //FIXME: this doesn't run!:
    queue.toStream().foreach { ff =&gt; ff foreach { ss =&gt;
      println(s"LENGTH IS: $ss")
    }}
  }
  val stringList = List("abc", "123", "abc123")
  val stringQueue = QueueStream[String]()
  stringList.foreach{ii =&gt; stringQueue.enqueue(ii)}
  val strlenQueue = QueueStream[Int]()
  var sz = strlenQueue.size()
  println(s"strlenQueue size is $sz")
  
  stringQueue.toStream().foreach{ ff =&gt; ff foreach { ss =&gt;
    strlenQueue.enqueue(ss.length)
  }}
  sz = strlenQueue.size()
  println(s"strlenQueue size is $sz")
  printIntsInQueue(strlenQueue)
  </code></pre>
  
  <p><strong>Edit: Working code</strong></p>
  
  <pre><code>import java.util.concurrent.{Executors, BlockingQueue, LinkedBlockingQueue}
  import scala.collection.JavaConversions._
  import scala.concurrent.{ExecutionContext, Future}
  import scala.concurrent.duration._
  import scala.language.postfixOps
  
  implicit val ec = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(8))
  
  class QueueStream[A]( private val queue: BlockingQueue[Option[A]] ) {
  
    implicit protected val defaultTimeout: Duration = 100 milliseconds
  
    def toStream(q: BlockingQueue[Option[A]] = queue ): Stream[A] =
      queue.take() match {
        case Some(a) =&gt; Stream cons(a, toStream())
        case None =&gt; Stream.empty
      }
    def toStreamNoWait(q: BlockingQueue[Option[A]] = queue ): Stream[A] = {
      val timeout = implicitly[Duration]
      queue.poll(timeout.toMillis, MILLISECONDS) match {
        case Some(a) =&gt; Stream cons(a, toStreamNoWait())
        case null =&gt; Stream.empty
        case None =&gt; Stream.empty
      }
    }
  
    def size () = queue.size()
    def close() = queue add None
    def enqueue( as: A* ) = queue addAll as.map( Some(_) )
  }
  object QueueStream {
    def apply[A]() = new QueueStream[A](new LinkedBlockingQueue)
  }
  
  def printIntsInQueue(queue: QueueStream[Int]): Unit = {
    println("About to print some Ints...")
    //FIXME: this doesn't run!:
    queue.toStream().foreach {ss =&gt;
      println(s"LENGTH IS: $ss")
    }
  }
  val stringList = List("abc", "123", "abc123")
  val stringQueue = QueueStream[String]()
  stringList.foreach{ii =&gt; stringQueue.enqueue(ii)}
  stringQueue.close()
  val strlenQueue = QueueStream[Int]()
  var sz = strlenQueue.size()
  println(s"strlenQueue size is $sz")
  stringQueue.toStream().foreach { ss =&gt;
    strlenQueue.enqueue(ss.length)
  }
  strlenQueue.close()
  sz = strlenQueue.size()
  println(s"strlenQueue size is $sz")
  printIntsInQueue(strlenQueue)
  </code></pre>