Add UnicastSubject

src/main/scala/rx/lang/scala/subjects/UnicastSubject.scala

@@ -0,0 +1,57 @@
+/**
+  * Copyright 2013 Netflix, Inc.
+  *
+  * Licensed under the Apache License, Version 2.0 (the "License");
+  * you may not use this file except in compliance with the License.
+  * You may obtain a copy of the License at
+  *
+  *     http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing, software
+  * distributed under the License is distributed on an "AS IS" BASIS,
+  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  * See the License for the specific language governing permissions and
+  * limitations under the License.
+  */
+package rx.lang.scala.subjects
+
+import rx.annotations.Experimental
+import rx.lang.scala.Subject
+
+/**
+  * $experimental A `Subject` variant which buffers events until a single `Subscriber` arrives and replays
+  * them to it and potentially switches to direct delivery once the `Subscriber` caught up and requested an
+  * unlimited amount. In this case, the buffered values are no longer retained. If the `Subscriber` requests
+  * a limited amount, queueing is involved and only those values are retained which weren't requested by the
+  * `Subscriber` at that time.
+  *
+  * @define experimental
+  * <span class="badge badge-red" style="float: right;">EXPERIMENTAL</span>
+  */
+@Experimental
+object UnicastSubject {
+
+  /**
+    * $experimental Constructs an empty `UnicastSubject` instance with the default capacity hint of 16 elements.
+    *
+    * @tparam T the input and output value type
+    * @return the created `UnicastSubject` instance
+    */
+  @Experimental
+  def apply[T](): UnicastSubject[T] = new UnicastSubject[T](rx.subjects.UnicastSubject.create[T]())
+
+  /**
+    * $experimental Constructs an empty UnicastSubject instance with a capacity hint.
+    * <p>The capacity hint determines the internal queue's island size: the larger
+    * it is the less frequent allocation will happen if there is no subscriber
+    * or the subscriber hasn't caught up.
+    *
+    * @param capacity the capacity hint for the internal queue
+    * @tparam T the input and output value type
+    * @return the created `UnicastSubject` instance
+    */
+  @Experimental
+  def apply[T](capacity: Int): UnicastSubject[T] = new UnicastSubject[T](rx.subjects.UnicastSubject.create(capacity))
+}
+
+private [scala] class UnicastSubject[T] private [scala] (val asJavaSubject: rx.subjects.UnicastSubject[T]) extends Subject[T] {}

src/test/scala/rx/lang/scala/SubjectTests.scala

@@ -15,20 +15,12 @@
  */
 package rx.lang.scala
 
-import org.junit.{Assert, Test}
-import org.scalatest.junit.JUnitSuite
-import scala.concurrent.duration._
 import scala.language.postfixOps
-import rx.lang.scala.schedulers.TestScheduler
-import rx.lang.scala.subjects.{AsyncSubject, ReplaySubject, BehaviorSubject}
-import org.mockito.Mockito._
-import org.mockito.Matchers._
-import org.junit.Assert.assertEquals
-import org.junit.Assert.assertTrue
-import org.junit.Assert.assertFalse
-import org.junit.Ignore
-import org.junit.Test
+
+import org.junit.Assert.{assertEquals, assertFalse, assertTrue}
+import org.junit.{Assert, Test}
 import org.scalatest.junit.JUnitSuite
+import rx.lang.scala.subjects.{AsyncSubject, BehaviorSubject, ReplaySubject, UnicastSubject}
 
 class SubjectTest extends JUnitSuite {
 
@@ -320,4 +312,49 @@ class SubjectTest extends JUnitSuite {
 
   }
 
-}
+  @Test def UnicastSubjectIsABuffer() {
+
+    val channel = UnicastSubject[Integer]
+    channel.onNext(42)
+
+    var lastA: Integer = null
+    var errorA, completedA: Boolean = false
+    val a = channel.subscribe(x => { lastA = x}, e => { errorA = true} , () => { completedA = true })
+
+    assertEquals(42, lastA)
+
+    channel.onNext(4711)
+    assertEquals(4711, lastA)
+
+    var lastB: Integer = null
+    var errorB, completedB: Boolean = false
+    val b = channel.subscribe(x => { lastB = x}, e => { errorB = true} , () => { completedB = true })
+
+    assertEquals(null, lastB)
+    assertFalse(completedB)
+    assertTrue(errorB) // only a single subscriber is allowed
+
+    val channel2 = UnicastSubject[Integer]
+    channel2.onNext(13)
+    channel2.onCompleted()
+
+    var lastC: Integer = null
+    var errorC, completedC: Boolean = false
+    val c = channel2.subscribe(x => { lastC = x}, e => { errorC = true} , () => { completedC = true })
+
+    assertEquals(13, lastC)
+    assertTrue(completedC)
+    assertFalse(errorC)
+
+    val channel3 = UnicastSubject[Integer]
+    channel3.onError(new Exception("Boom"))
+
+    var lastD: Integer = null
+    var errorD, completedD: Boolean = false
+    val d = channel3.subscribe(x => { lastD = x}, e => { errorD = true} , () => { completedD = true })
+
+    assertEquals(null, lastD)
+    assertFalse(completedD)
+    assertTrue(errorD)
+  }
+}