title: Buffering of grouped items with RxJava
body: <p>I'm trying to implement a stream handling with RxJava.
  I want to handle different steps.</p>
  
  <ul>
  <li>I start with a publish subject where sensors data are pushed</li>
  <li>I want to group these sensors data by their sensor type (an identifier)</li>
  <li>For each group I want to buffer these sensors data</li>
  <li>When the buffer is full or times out, I want to compute an average from all sensors value in that group</li>
  <li>And finally I want to re-join all these groups into one single output stream</li>
  </ul>
  
  <p>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. </p>
  
  <pre><code>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&lt;Sensor&gt; publishSubject = PublishSubject.create();
  
      static class Sensor {
          int type;
          float value;
  
          Sensor(int type, float value) {
              this.type = type;
              this.value = value;
          }
      }
  
      private PublishSubject&lt;Sensor&gt; listenSensors() {
          return publishSubject;
      }
  
      private static Sensor getValueAverage(List&lt;Sensor&gt; sensors) {
          int count = sensors.size();
          float total = sensors.stream().map(sensor -&gt; 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&lt;Sensor&gt; sensorPublishSubject) throws InterruptedException {
  
          new Thread(() -&gt; {
              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&lt;List&lt;Sensor&gt;&gt; flatpMapSensor(List&lt;Sensor&gt; sensors) {
          return Observable
              .fromIterable(sensors)
              .groupBy(s -&gt; s.type)
              .flatMapSingle(Observable::toList);
      }
  
      // Testing code
      static public void main(String args[]) throws InterruptedException {
          Main main = new Main();
          main.listenSensors()
              .publish(p -&gt; p
                  .buffer(20, TimeUnit.SECONDS, 10)
                  .filter(list -&gt; !list.isEmpty()))
              .flatMap(Main::flatpMapSensor)
              .map(Main::getValueAverage)
              .subscribe(sensor -&gt; System.out.println("AVG " + getStringType(sensor.type) + " " + sensor.value));
          emitRandomValue(main.publishSubject);
  
          Thread.sleep(90000);
  
      }
  }
  </code></pre>
  
  <p>So my question is : How can I have a separate buffer for each sensor type ?</p>