Reactive aggregation

11

I got a request to share an example of how to do an aggregation with cursor.observe

@brucejo showed an interesting example using reactive-publish, but be aware, that this runs the aggregation for every user that subscribes this publication and on every change.

if the result is the same for every user, you aggregate too much. But it’s an interesting way, if you aggregate something per user.

So the other way with cursor.observe would be something like this:


const doTheAggregation () => {
  // here you do the actual aggregation. In this example its one job
  const data1 = Data1.find(...).fetch();
  const data2 = Data2.find(...).fetch();
  // ...
  AggregatedCollection.update(...);
}

// this function will update the AggregatedCollection whenever it is run. 
// But as it might be expensive, we want to have control when it should run
// so our approach is to run it whenever one of the source data changes:

Data1.find({}).observe({
   added: doTheAggregation,
   changed: doTheAggregation,
   removed: doTheAggregation
});

Data2.find({}).observe({
   added: doTheAggregation,
   changed: doTheAggregation,
   removed: doTheAggregation
});

// be aware that added is also called for every document whenever the server starts
// you could do something like this to prevent this:

let startup = true;
Data1.find({}).observe({
   added: () => {
     if(!startup) {
       doTheAggregation();
     } 
  },
   changed: doTheAggregation,
   removed: doTheAggregation
});
startup = false;


// also you might want to throttle or debounce doTheAggregation with _.debounce or _.throttle so that it does not run when multiple documents are changed in a short time period.
1 Like