It works!

But it's probably super hacky. I know I'm blocking up some threads, so I
need to clean that up really bad

src/main/kotlin/is/kow/deskscreen/ticker/TickerView.kt

@@ -2,9 +2,12 @@ package `is`.kow.deskscreen.ticker
 
 import com.github.thomasnield.rxkotlinfx.changes
 import com.github.thomasnield.rxkotlinfx.doOnNextFx
+import io.reactivex.BackpressureStrategy
+import io.reactivex.Flowable
 import io.reactivex.Observable
+import io.reactivex.processors.PublishProcessor
 import io.reactivex.rxjavafx.sources.Flag
-import io.reactivex.subjects.PublishSubject
+import io.reactivex.schedulers.Schedulers
 import javafx.animation.Animation
 import javafx.animation.KeyFrame
 import javafx.animation.Timeline
@@ -12,14 +15,12 @@ import javafx.beans.property.SimpleIntegerProperty
 import javafx.collections.FXCollections
 import javafx.event.ActionEvent
 import javafx.event.EventHandler
-import javafx.scene.Node
 import javafx.scene.layout.Pane
 import javafx.scene.shape.Rectangle
 import javafx.scene.text.Text
 import javafx.util.Duration
 import mu.KotlinLogging
 import tornadofx.*
-import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
 
@@ -31,7 +32,8 @@ class TickerEntry(
     //Number of pixels from the Right side of the screen
     //Will be updated by the entry thing, to render stuff
     val xPosition = SimpleIntegerProperty(this, "xPosition", 0)
-    override fun toString():String {
+
+    override fun toString(): String {
         return "TickerEntry(title=$title, content=$content)"
     }
 }
@@ -106,7 +108,7 @@ class MarqueeView(inside: Pane) : View() {
     private val logger = KotlinLogging.logger {}
     private val OFFSET = 5.0 //Amount of space between entries!
 
-    private val tickEntries = PublishSubject.create<TickerEntry>()
+    private val tickEntries = PublishProcessor.create<TickerEntry>()
 
     data class ClearedTick(val entry: TickerEntry)
 
@@ -126,55 +128,70 @@ class MarqueeView(inside: Pane) : View() {
         root.clip = rectangle
 
         //Hook up an observable looking for the right kind of change to the list
-        tickerModel.tickets.changes()
-                .filter {
-                    it.flag == Flag.ADDED
-                }
-                .map {
-                    it.value
-                }
-                .doOnNext {
-                    logger.debug("Received addition to tickerModel!: $it")
-
-                    //Queue it up to be processed
-                    tickEntries.onNext(it)
-                    //start an animation for this entry
-
-                    //Need to somehow queue up the entry, so it doesn't fire until the other one is done, or something...
-                    //Some kind of state tracking for the current item in progress, find it's end layout, if it's greater than the width, start there, otherwise start at width
-                }
-                .subscribe()
+//        tickerModel.tickets.changes()
+//                .filter {
+//                    it.flag == Flag.ADDED
+//                }
+//                .map {
+//                    it.value
+//                }
+//                .doOnNext {
+//                    logger.debug("Received addition to tickerModel!: $it")
+//
+//                    //Queue it up to be processed
+//                    tickEntries.onNext(it)
+//                    //start an animation for this entry
+//
+//                    //Need to somehow queue up the entry, so it doesn't fire until the other one is done, or something...
+//                    //Some kind of state tracking for the current item in progress, find it's end layout, if it's greater than the width, start there, otherwise start at width
+//                }
+//                .subscribe()
 
 
-        val ready = AtomicBoolean(true)
         //Is this being take based, or push based. probably push based...
         //It isn't delaying the incoming new ones... they're getting published immediately.
-        tickEntries
-                .takeWhile {
-                    ready.get()
-                }
+
+        Flowable.create<TickerEntry>({ emitter ->
+            tickerModel.tickets.changes()
+                    .observeOn(Schedulers.computation())
+                    .filter {
+                        it.flag == Flag.ADDED
+                    }
+                    .map {
+                        it.value
+                    }
+                    .doOnNext {
+                        emitter.onNext(it)
+                    }
+                    .subscribe()
+            //There isn't really a cancellation thing...
+        },
+                BackpressureStrategy.MISSING)
+                .onBackpressureBuffer(1024)
                 .doOnNext {
-                    logger.debug("DOING A THING WITH $it") //Why did it skip my second one?
+                    logger.debug("Backpressure driven work for $it")
                 }
-                .flatMap {
+                .map {
                     //start an animation, but don't release another until it's past a point
                     //Except cannot.... or consume a blocking, but that's gross, probably easier tho....
                     //do some kind of callback thing like for steam
-                    val clearedSubject = PublishSubject.create<ClearedTick>()
+                    val clearedSubject = PublishProcessor.create<ClearedTick>()
+                    Pair(it, clearedSubject)
+                }
+                .doOnNextFx {
                     //This will fire off the request to animate the entry, and not move forward until the next thing passes. I think
-                    animateText(it, clearedSubject)
-                    ready.set(false) // not ready!
-                    clearedSubject
+                    animateText(it.first, it.second)
+                }
+                .map {
+                    it.second.blockingFirst() //This part I have to figure out super bad...
                 }
                 .doOnNext {
                     // This emits properly when it's done, which is awesome, but it's not blocking upstream...
                     logger.info("$it CLEARED!")
-                    ready.set(true) // ready again! Thought this would work, but it doesn't
                 }
                 .doOnComplete {
                     logger.info("TICK ENTRIES COMPLETED, retrying for another entry!")
                 }
-                .repeat() //Should rpobably have a delay?
                 .subscribe()
     }
 
@@ -190,7 +207,7 @@ class MarqueeView(inside: Pane) : View() {
     }
 
     //Will need to fire one of these for each thing, with some space detection, based on when one is ending.
-    private fun animateText(entry: TickerEntry,  publishSubject: PublishSubject<ClearedTick>) {
+    private fun animateText(entry: TickerEntry, publishSubject: PublishProcessor<ClearedTick>) {
 
         val text = createText(entry.content)
         val emptyStart = root.prefWidth //It's weird, like a race condition
@@ -209,7 +226,7 @@ class MarqueeView(inside: Pane) : View() {
 
             //instead I want to move completely from the right all the way off until it's gone
             //end state of this text is the entire width of the text off the right so layoutX <= 0 - textWidth - offset
-            if(textWidth + layoutX  + OFFSET <= paneWidth && !cleared.get()) {
+            if (textWidth + layoutX + OFFSET <= paneWidth && !cleared.get()) {
                 //There's enough room to start another, how can I notify this?
                 publishSubject.onNext(ClearedTick(entry))
                 cleared.set(true)