src/lib/pipeline.js
/**
 * Group events according to a window strategy and apply a reducer to each window.
 *
 * If the stream is split (`by` was called), maintain a different window for each key.
 */
Group events according to a window strategy and apply a reducer to each window.
If the stream is split (`by` was called), maintain a different window for each key.
Group events according to a window strategy and apply a reducer to each window . If the stream is split ( ` by ` was called ) maintain a different window for each key .
function window({
  strategy,
  reducer,
  allowedLateness,
  fireEvery,
  watermarkDelay,
}) {
  const id = nextId();
  return stream => {
    return event => {
      const t = event.get('time');
      const watermark = now() - watermarkDelay;
      instruments.gauge('pipeline.window.delta', t - now());
      if (t < watermark - allowedLateness) {
        console.log('WARNING', 'Dropping late event.');
        console.log(
          'Watermark:',
          iso(watermark),
          'Event time:',
          iso(t),
          'Delta:',
          t - watermark
        );
        return;
      }
      windows = windows.updateIn([id, event.get('key')], Map(), windows => {
        let eventWindows = strategy.assign(event);
        windows = eventWindows.reduce((windows, window) => {
          return windows.update(
            window.get('start') + ':' + window.get('end'),
            window.set('acc', reducer.init()),
            w => {
              return w
                .update('acc', acc => reducer.step(acc, event.get('data')))
                .set('triggered', false);
            }
          );
        }, windows);
        windows = windows.map(window => {
          if (
            fireEvery &&
            watermark < window.get('end') &&
            watermark - window.get('lastFired', window.get('start')) >=
              fireEvery
          ) {
            forward(
              stream,
              Map({
                time: window.get('start'),
                data: reducer.result(window.get('acc')),
                key: event.get('key'),
              })
            );
            return window.set('lastFired', watermark);
          }
          if (window.get('end') <= watermark && !window.get('triggered')) {
            forward(
              stream,
              Map({
                time: window.get('start'),
                data: reducer.result(window.get('acc')),
                key: event.get('key'),
              })
            );
            return window.set('triggered', true);
          }
          return window;
        });
        return windows.filter(
          window => window.get('end') >= watermark - allowedLateness
        );
      });
    };
  };
}
function window ( { strategy , reducer , allowedLateness , fireEvery , watermarkDelay , } ) { const id = nextId ( ) ; return stream => { return event => { const t = event . get ( 'time' ) ; const watermark = now ( ) - watermarkDelay ; instruments . gauge ( 'pipeline.window.delta' , t - now ( ) ) ; if ( t < watermark - allowedLateness ) { console . log ( 'WARNING' , 'Dropping late event.' ) ; console . log ( 'Watermark:' , iso ( watermark ) , 'Event time:' , iso ( t ) , 'Delta:' , t - watermark ) ; return ; } windows = windows . updateIn ( [ id , event . get ( 'key' ) ] , Map ( ) , windows => { let eventWindows = strategy . assign ( event ) ; windows = eventWindows . reduce ( ( windows , window ) => { return windows . update ( window . get ( 'start' ) + ':' + window . get ( 'end' ) , window . set ( 'acc' , reducer . init ( ) ) , w => { return w . update ( 'acc' , acc => reducer . step ( acc , event . get ( 'data' ) ) ) . set ( 'triggered' , false ) ; } ) ; } , windows ) ; windows = windows . map ( window => { if ( fireEvery && watermark < window . get ( 'end' ) && watermark - window . get ( 'lastFired' , window . get ( 'start' ) ) >= fireEvery ) { forward ( stream , Map ( { time : window . get ( 'start' ) , data : reducer . result ( window . get ( 'acc' ) ) , key : event . get ( 'key' ) , } ) ) ; return window . set ( 'lastFired' , watermark ) ; } if ( window . get ( 'end' ) <= watermark && ! window . get ( 'triggered' ) ) { forward ( stream , Map ( { time : window . get ( 'start' ) , data : reducer . result ( window . get ( 'acc' ) ) , key : event . get ( 'key' ) , } ) ) ; return window . set ( 'triggered' , true ) ; } return window ; } ) ; return windows . filter ( window => window . get ( 'end' ) >= watermark - allowedLateness ) ; } ) ; } ; } ; }
Group events according to a window strategy and apply a reducer to each window.
Group events according to a window strategy and apply a reducer to each window .
// Drop late events // assign event to windows // update windows // trigger windows (FIXME: Move this logic out of this function) // Garbage collect windows