code: @Override
      public Subscriber<? super CellValue> call(final Subscriber<? super CellValue> child) {
  
          Subscriber<CellValue> parent = Subscribers.from(new Observer<CellValue>() {
  
              @Override
              public void onCompleted() {
                  // MapDb.INSTANCE.getDb().commit();
                  try {
                      log.info("starting to emit map values");
                      synchronized (map) {
                          for (Entry<Position, Double> entry : map.entrySet()) {
                              CellValue cv = new CellValue(entry.getKey().lat(),
                                      entry.getKey().lon(), entry.getValue().doubleValue());
                              child.onNext(cv);
                          }
                      }
                      child.onCompleted();
                  } catch (Throwable t) {
                      onError(t);
                  }
  
              }
  
              @Override
              public void onError(Throwable e) {
                  child.onError(e);
              }
  
              @Override
              public void onNext(CellValue cv) {
                  Position position = new Position((float) cv.getCentreLat(), (float) cv
                          .getCentreLon());
                  Double val = map.putIfAbsent(position, cv.getValue());
                  if (val != null)
                      map.put(position, val + cv.getValue());
              }
          });
          child.add(parent);
          return parent;
      }
docs: private final AtomicLong count = new AtomicLong();