Add intersperse operator

AUTHORS

@@ -17,4 +17,7 @@ Jisoo Park
 https://github.com/guersam
 
 Dawid Dworak
-https://github.com/ddworak
+https://github.com/ddworak
+
+Omar Mainegra
+https://github.com/omainegra

monix-reactive/shared/src/main/scala/monix/reactive/internal/operators/IntersperseObservable.scala

@@ -0,0 +1,72 @@
+/*
+ * Copyright (c) 2014-2016 by its authors. Some rights reserved.
+ * See the project homepage at: https://monix.io
+ *
+ * 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 monix.reactive.internal.operators
+
+import monix.execution.Ack.Continue
+import monix.execution.{Ack, Cancelable}
+import monix.reactive.Observable
+import monix.reactive.observers.Subscriber
+
+import scala.concurrent.Future
+
+private[reactive] final class IntersperseObservable[+A](
+  source: Observable[A],
+  start: Option[A],
+  separator: A,
+  end: Option[A] ) extends Observable[A] { self =>
+
+  override def unsafeSubscribeFn(out: Subscriber[A]): Cancelable = {
+    val upstream = source.unsafeSubscribeFn(new Subscriber[A] {
+      implicit val scheduler = out.scheduler
+
+      private[this] var firstTime = true
+      private[this] var downstreamAck = Continue : Future[Ack]
+
+      override def onNext(elem: A): Future[Ack] = {
+        downstreamAck = if (firstTime) {
+          firstTime = false
+          start.map(out.onNext).getOrElse(Continue).syncFlatMap {
+            case Continue => out.onNext(elem)
+            case ack => ack
+          }
+        }
+        else {
+          out.onNext(separator).syncFlatMap {
+            case Continue => out.onNext(elem)
+            case ack => ack
+          }
+        }
+
+        downstreamAck
+      }
+
+      def onError(ex: Throwable) = out.onError(ex)
+
+      def onComplete() = {
+        downstreamAck.syncOnContinue {
+          if (end.nonEmpty) out.onNext(end.get)
+          out.onComplete()
+        }
+      }
+    })
+
+    Cancelable { () =>
+      upstream.cancel()
+    }
+  }
+}

monix-reactive/shared/src/main/scala/monix/reactive/observables/ObservableLike.scala

@@ -2183,6 +2183,24 @@ trait ObservableLike[+A, Self[+T] <: ObservableLike[T, Self]]
   /** Zips the emitted elements of the source with their indices. */
   def zipWithIndex: Self[(A, Long)] =
     self.liftByOperator(new ZipWithIndexOperator[A])
+
+  /** Creates a new observable from this observable that will emit a specific `separator`
+    * between every pair of elements.
+    *
+    * @param separator is the separator
+    */
+  def intersperse[B >: A](separator: B): Self[B] =
+    self.transform(self => new IntersperseObservable(self, None, separator, None))
+
+  /** Creates a new observable from this observable that will emit the `start` element
+    * followed by the upstream elements paired with the `separator`, and lastly the `end` element.
+    *
+    * @param start is the first element emitted
+    * @param separator is the separator
+    * @param end the last element emitted
+    */
+  def intersperse[B >: A](start: B, separator: B, end: B): Self[B] =
+    self.transform(self => new IntersperseObservable(self, Some(start), separator, Some(end)))
 }
 
 object ObservableLike {

monix-reactive/shared/src/test/scala/monix/reactive/internal/operators/IntersperseSuite.scala

@@ -0,0 +1,119 @@
+/*
+ * Copyright (c) 2014-2016 by its authors. Some rights reserved.
+ * See the project homepage at: https://monix.io
+ *
+ * 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 monix.reactive.internal.operators
+
+import monix.execution.Ack.Continue
+import monix.reactive.subjects.PublishSubject
+import monix.reactive.{Observable, Observer}
+
+import scala.concurrent.duration._
+
+object IntersperseSuite extends BaseOperatorSuite {
+  def createObservable(sourceCount: Int) = Option {
+    val sample = Observable.range(0, sourceCount).intersperse(1L)
+    Sample(sample, count(sourceCount), sum(sourceCount), 0.seconds, 0.seconds)
+  }
+
+  def count(sourceCount: Int) = sourceCount*2 - 1
+  def sum(sourceCount: Int) = (sourceCount*(sourceCount - 1))/2 + sourceCount - 1
+
+  def observableInError(sourceCount: Int, ex: Throwable) = Some {
+    val sample = createObservableEndingInError(Observable.range(0, sourceCount), ex).intersperse(0L)
+    Sample(sample, count(sourceCount), sum(sourceCount), 0.seconds, 0.seconds)
+  }
+
+  override def cancelableObservables(): Seq[Sample] = {
+    val sample = Observable.range(0, 10).delayOnNext(1.second).intersperse(0L)
+    Seq(Sample(sample, 0, 0, 0.seconds, 0.seconds))
+  }
+
+  def brokenUserCodeObservable(sourceCount: Int, ex: Throwable) = None
+
+  test("start is the first emitted element") { implicit s =>
+    val obs = PublishSubject[Int]()
+
+    var received = Vector.empty[Int]
+    var wasCompleted = false
+
+    obs.intersperse(start = -1, separator = -2, end = -3).subscribe(new Observer[Int] {
+      def onNext(elem: Int) = {
+        received :+= elem
+        Continue
+      }
+
+      def onError(ex: Throwable) = ()
+      def onComplete() = wasCompleted = true
+    })
+
+    obs.onNext(1); s.tick()
+    assertEquals(received.headOption, Some(-1))
+
+    obs.onComplete(); s.tick()
+    assert(wasCompleted)
+  }
+
+  test("end is the last emitted element") { implicit s =>
+    val obs = PublishSubject[Int]()
+
+    var received = Vector.empty[Int]
+    var wasCompleted = false
+
+    obs.intersperse(start = -1, separator = -2, end = -3).unsafeSubscribeFn(new Observer[Int] {
+      def onNext(elem: Int) = {
+        received :+= elem
+        Continue
+      }
+
+      def onError(ex: Throwable) = ()
+      def onComplete() = wasCompleted = true
+    })
+
+    obs.onNext(1); s.tick()
+    obs.onNext(2); s.tick()
+    obs.onComplete(); s.tick()
+    assertEquals(received.lastOption, Some(-3))
+    assert(wasCompleted)
+  }
+
+  test("separator is paired with emitted elements") { implicit s =>
+    val obs = PublishSubject[Int]()
+
+    var received = Vector.empty[Int]
+    var wasCompleted = false
+
+    obs.intersperse(-1,0,-1).unsafeSubscribeFn(new Observer[Int] {
+      def onNext(elem: Int) = {
+        received :+= elem
+        Continue
+      }
+
+      def onError(ex: Throwable) = ()
+      def onComplete() = wasCompleted = true
+    })
+
+    obs.onNext(1); s.tick()
+    obs.onNext(2); s.tick()
+    obs.onNext(3); s.tick()
+    obs.onNext(4); s.tick()
+    obs.onNext(5); s.tick()
+    obs.onComplete(); s.tick()
+
+    assertEquals(received, Vector(-1,1,0,2,0,3,0,4,0,5,-1))
+    assert(wasCompleted)
+  }
+}