Buffering of grouped items with RxJava

Question

I'm trying to implement a stream handling with RxJava. I want to handle different steps.

• I start with a publish subject where sensors data are pushed
• I want to group these sensors data by their sensor type (an identifier)
• For each group I want to buffer these sensors data
• When the buffer is full or times out, I want to compute an average from all sensors value in that group
• And finally I want to re-join all these groups into one single output stream

So far with the code example below, I have a buffer that is shared for all sensors data. I don't understand how to create a buffer for each group and then do my computation. As I'm new to RxJava, I don't understand all the concepts and I'm stuck with my issue.

import io.reactivex.Observable;
import io.reactivex.subjects.PublishSubject;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class Main {

    private static final int SENSOR_TEMPERATURE = 1;
    private static final int SENSOR_HUMIDITY = 2;

    private PublishSubject<Sensor> publishSubject = PublishSubject.create();

    static class Sensor {
        int type;
        float value;

        Sensor(int type, float value) {
            this.type = type;
            this.value = value;
        }
    }

    private PublishSubject<Sensor> listenSensors() {
        return publishSubject;
    }

    private static Sensor getValueAverage(List<Sensor> sensors) {
        int count = sensors.size();
        float total = sensors.stream().map(sensor -> sensor.value).reduce(Float::sum).orElse(0f);
        float avg = total / count;
        return new Sensor(sensors.get(0).type, avg);
    }

    //Map type
    private static String getStringType(int type) {
        if (type == SENSOR_HUMIDITY) {
            return "HUMIDITY";
        }
        else if (type == SENSOR_TEMPERATURE) {
            return "TEMPERATURE";
        }
        return "OTHER";
    }

    private static void emitRandomValue(PublishSubject<Sensor> sensorPublishSubject) throws InterruptedException {

        new Thread(() -> {
            int randomDelay = 0;

            while (true) {
                int randomType = (int) ((Math.random() * 10 % 2) + 1);
                randomDelay = (int) (Math.random() * 3000);
                float randomValue = (float) (Math.random() * 100);
                System.out.println("EMIT: " + getStringType(randomType) + " " + randomValue);
                sensorPublishSubject.onNext(new Sensor(randomType, randomValue));
                try {
                    Thread.sleep(randomDelay);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

        }).start();
    }
    static Observable<List<Sensor>> flatpMapSensor(List<Sensor> sensors) {
        return Observable
            .fromIterable(sensors)
            .groupBy(s -> s.type)
            .flatMapSingle(Observable::toList);
    }

    // Testing code
    static public void main(String args[]) throws InterruptedException {
        Main main = new Main();
        main.listenSensors()
            .publish(p -> p
                .buffer(20, TimeUnit.SECONDS, 10)
                .filter(list -> !list.isEmpty()))
            .flatMap(Main::flatpMapSensor)
            .map(Main::getValueAverage)
            .subscribe(sensor -> System.out.println("AVG " + getStringType(sensor.type) + " " + sensor.value));
        emitRandomValue(main.publishSubject);

        Thread.sleep(90000);

    }
}

So my question is : How can I have a separate buffer for each sensor type ?

Answer 1

What if you shift the buffer() and groupBy() calls?

static public void main(String args[]) throws InterruptedException {
   Main main = new Main();
   main.listenSensors()
       .groupBy(s -> s.type) // group by type
       .flatMap(l -> l.buffer(20, SECONDS, 10).map(Main::getValueAverage)) // buffer groups by type and compute the average
       .subscribe(sensor -> System.out.println("AVG " + getStringType(sensor.type) + " " + sensor.value));
   emitRandomValue(main.publishSubject);

   Thread.sleep(90000);
}