Menu
I have an observable of events: ElementAdded (A), ElementRemoved (R), ActionStarted (S) and ActionFinished (F). Some of the Adds and Removes are sandwiched between an ActionStarted and an ActionFinished. I want to replace that subsequence of events with a single event ElementMoved (M) while letting the non-sandwiched events fly without delay. The ElementMoved events should contain an array with all the events it is replacing. Here is an example:
---A--A--R--S-A-R-F-R-A-A--
(my transformation)
---A--A--R--------M-R-A-A--
ElementMoved should be appear the moment the ActionFinished event is fired.
Additionally, if no ActionFinished event is fired after a timeout T since the last sandwiched event, then all original events should fire instead:
-----T
---A1--A2--R3--S4-A5-R6------------R7-A8-A9--
(my transformation)
---A1--A2--R3---------------S4A5R6-R7-A8-A9--
There could be an ActionFinished event that is fired after the timeout or it could never happen (like in the example). If it never happens, there is nothing to do. It it happens and there is NO window open, the ActionFinished event to make it into the new stream by itself. For example:
-----T
---A1--A2--R3--S4-A5-R6------------F7-A8-A9--
(my transformation)
---A1--A2--R3---------------S4A5R6-F7-A8-A9--
Basically, if the transformation is not able to close a window in a given timeout, it should flush all the withheld events untouched.
This flushing of events should also happen if a new S event is fired before a corresponding F event. (This new S event should be withheld as per the logic above). For instance
---A1--A2--R3--S4-A5-R6--S7---R9-A9-A10-F11-A12--
(my transformation)
---A1--A2--R3------------S4A5R6---------M7- A12--
I've been playing with the window operator for a while with no luck. The buffer operator introduces a delay for free-floating events, which is not acceptable in my case, the. Scan emits as many events as the original stream, which is not what I want. I'm certainly lost so any help would be very much appreciated.
Edit 1: Added case about flushing when a new S event appears while a window is open
Edit 2: Clarify that the Move events should contain the list of events it is replacing.
Edit 3: Changed tag from rx-java to rx-java2
Edit 4: Clarify what happens if the ActionFinished event comes after the timeout kicks in.
Thank you!
712
Since my last answer was deleted by "reviewers", here is the answer again with the full source code. If this gets deleted because the long code parts, I don't know what to do. Note that the OP's question requires a complicated operator:
package hu.akarnokd.rxjava;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import io.reactivex.internal.util.BackpressureHelper;
import org.reactivestreams.*;
import io.reactivex.*;
import io.reactivex.Scheduler.Worker;
import io.reactivex.disposables.*;
import io.reactivex.schedulers.Schedulers;
public class Main {
public static void main(String[] args) {
Flowable<String> source = Flowable.just(
"A", "A", "R", "S", "A", "R", "F", "R", "A", "A");
source.lift(new ConditionalCompactor(
500, TimeUnit.SECONDS, Schedulers.computation()))
.subscribe(System.out::println, Throwable::printStackTrace);
}
static final class ConditionalCompactor implements FlowableOperator<String, String> {
final Scheduler scheduler;
final long timeout;
final TimeUnit unit;
ConditionalCompactor(long timeout, TimeUnit unit,
Scheduler scheduler) {
this.scheduler = scheduler;
this.timeout = timeout;
this.unit = unit;
}
@Override
public Subscriber<? super String> apply(Subscriber<? super String> t) {
return new ConditionalCompactorSubscriber(
t, timeout, unit, scheduler.createWorker());
}
static final class ConditionalCompactorSubscriber
implements Subscriber<String>, Subscription {
final Subscriber<? super String> actual;
final Worker worker;
final long timeout;
final TimeUnit unit;
final AtomicInteger wip;
final SerialDisposable mas;
final Queue<String> queue;
final List<String> batch;
final AtomicLong requested;
Subscription s;
static final Disposable NO_TIMER;
static {
NO_TIMER = Disposables.empty();
NO_TIMER.dispose();
}
volatile boolean done;
Throwable error;
boolean compacting;
int lastLength;
ConditionalCompactorSubscriber(Subscriber<? super String> actual,
long timeout, TimeUnit unit, Worker worker) {
this.actual = actual;
this.worker = worker;
this.timeout = timeout;
this.unit = unit;
this.batch = new ArrayList<>();
this.wip = new AtomicInteger();
this.mas = new SerialDisposable();
this.mas.set(NO_TIMER);
this.queue = new ConcurrentLinkedQueue<>();
this.requested = new AtomicLong();
}
@Override
public void onSubscribe(Subscription s) {
this.s = s;
actual.onSubscribe(this);
}
@Override
public void onNext(String t) {
queue.offer(t);
drain();
}
@Override
public void onError(Throwable e) {
error = e;
done = true;
drain();
}
@Override
public void onComplete() {
done = true;
drain();
}
@Override
public void cancel() {
s.cancel();
worker.dispose();
}
@Override
public void request(long n) {
BackpressureHelper.add(requested, n);
s.request(n);
drain();
}
void drain() {
if (wip.getAndIncrement() != 0) {
return;
}
int missed = 1;
for (;;) {
long r = requested.get();
long e = 0L;
while (e != r) {
boolean d = done;
if (d && error != null) {
queue.clear();
actual.onError(error);
worker.dispose();
return;
}
String s = queue.peek();
if (s == null) {
if (d) {
actual.onComplete();
worker.dispose();
return;
}
break;
}
if (compacting) {
batch.clear();
batch.addAll(queue);
int n = batch.size();
String last = batch.get(n - 1);
if ("S".equals(last)) {
if (n > 1) {
actual.onNext(queue.poll());
mas.set(NO_TIMER);
lastLength = -1;
compacting = false;
e++;
continue;
}
// keep the last as the start of the new
if (lastLength <= 0) {
lastLength = 1;
mas.set(worker.schedule(() -> {
queue.offer("T");
drain();
}, timeout, unit));
this.s.request(1);
}
break;
} else
if ("T".equals(last)) {
actual.onNext(queue.poll());
compacting = false;
mas.set(NO_TIMER);
lastLength = -1;
e++;
continue;
} else
if ("F".equals(last)) {
actual.onNext("M");
while (n-- != 0) {
queue.poll();
}
compacting = false;
mas.set(NO_TIMER);
lastLength = -1;
e++;
} else {
if (lastLength != n) {
lastLength = n;
mas.set(worker.schedule(() -> {
queue.offer("T");
drain();
}, timeout, unit));
this.s.request(1);
}
break;
}
} else {
if ("A".equals(s) || "F".equals(s) || "R".equals(s)) {
queue.poll();
actual.onNext(s);
e++;
} else
if ("T".equals(s)) {
// ignore timeout markers outside the compacting mode
queue.poll();
} else {
compacting = true;
}
}
}
if (e != 0L) {
BackpressureHelper.produced(requested, e);
}
if (e == r) {
if (done) {
if (error != null) {
queue.clear();
actual.onError(error);
worker.dispose();
return;
}
if (queue.isEmpty()) {
actual.onComplete();
worker.dispose();
return;
}
}
}
missed = wip.addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
}
}
}
The operator's pattern is of a typical queue-drain but the drain phase contains the logic for combining certain subsequent patterns that also requires a different operation mode.
Edit updated to RxJava 2.
Edit 2 updated with backpressure support.
185
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With
